Skip to content

mcp-oauth.mdx

Latest commit

 

History

History
78 lines (55 loc) · 2.61 KB

mcp-oauth.mdx

File metadata and controls

78 lines (55 loc) · 2.61 KB
title description
Authorize MCP with OAuth
Authorize the user to delegate their identity to the MCP server using OAuth

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

Make sure `mcp` is installed as a dependency in your project. Import the necessary components from the Agent Stack SDK OAuth extension. Inject the OAuth extension into your agent function using the `Annotated` type hint. 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.