Skip to main content
Some MCP servers require OAuth authentication to access third-party services on behalf of users. The Agent Stack provides an OAuth extension that allows your agent to seamlessly handle OAuth flows through the user interface. The OAuth extension allows users to authorize your agent to access MCP servers that require authentication. The Agent Stack UI serves as a secure channel for users to perform the OAuth flow and grant access to third-party services through the MCP server. Once authorized, the OAuth extension automatically manages authentication headers, making it easy to use authenticated MCP clients in your agents.

​
Quickstart

1

Install the MCP dependency

Make sure mcp is installed as a dependency in your project.
2

Import the OAuth extension

Import the necessary components from the Agent Stack SDK OAuth extension.
3

Add OAuth parameter to your agent

Inject the OAuth extension into your agent function using the Annotated type hint.
4

Create authenticated MCP client

Use create_httpx_auth() to create an authentication handler for your MCP HTTP client.

​
Basic OAuth Example

Here’s how to build an agent that uses OAuth to authenticate with an MCP server: 
import os
from typing import Annotated

import pydantic
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

from agentstack_sdk.a2a.extensions.auth.oauth import OAuthExtensionServer, OAuthExtensionSpec
from agentstack_sdk.server import Server

server = Server()


@server.agent()
async def oauth_agent(
    oauth: Annotated[OAuthExtensionServer, OAuthExtensionSpec.single_demand()],
):
    """Agent that uses OAuth to authenticate with an MCP server"""
    mcp_url = "https://mcp.stripe.com"

    async with streamablehttp_client(
        url=mcp_url,
        auth=await oauth.create_httpx_auth(resource_url=pydantic.AnyUrl(mcp_url)) if oauth else None
    ) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool("get_stripe_account_info")
            yield result.content[0].text


def run():
    server.run(host=os.getenv("HOST", "127.0.0.1"), port=int(os.getenv("PORT", "8000")))


if __name__ == "__main__":
    run()
Always check if the OAuth extension is available before using it to comply with plain A2A clients that may not support OAuth authentication.