add example of a server protected by oauth (#578)

Author: roman-van-der-krogt

docs/configuration.mdx

@@ -136,9 +136,8 @@ authorization:
   issuer_url: https://auth.example.com
   resource_server_url: https://agent.example.com/mcp
   required_scopes: ["mcp.read", "mcp.write"]
-  introspection_endpoint: https://auth.example.com/oauth/introspect
-  introspection_client_id: ${INTROSPECTION_CLIENT_ID}
-  introspection_client_secret: ${INTROSPECTION_CLIENT_SECRET}
+  client_id: ${INTROSPECTION_CLIENT_ID}
+  client_secret: ${INTROSPECTION_CLIENT_SECRET}
 
 oauth:
   callback_base_url: https://agent.example.com

examples/oauth/protected_by_oauth/README.md

@@ -0,0 +1,81 @@
+# OAuth protected resource example
+
+This example shows how to integrate OAuth2 authentication to protect your MCP.
+
+## 1. App set up
+
+First, clone the repo and navigate to the functions example:
+
+```bash
+git clone https://github.com/lastmile-ai/mcp-agent.git
+cd mcp-agent/examples/oauth/protected_by_oauth
+```
+
+Install `uv` (if you don’t have it):
+
+```bash
+pip install uv
+```
+
+Sync `mcp-agent` project dependencies:
+
+```bash
+uv sync
+```
+
+## 2. Client registration
+To protect your MCP with OAuth2, you first need to register your application with an OAuth2 provider, as MCP follows the Dynamic Client Registration Protocol.
+You can configure either your own OAuth2 server, or use the one provided by MCP Agent Cloud (https://auth.mcp-agent.com).
+
+If you do not have a client registered already, you can use the `registration.py` script provided with this example.
+At the top of the file,
+1. update the URL for your authentication server,
+2. set the redirect URIs to point to your MCP endpoint (e.g. `https://your-mcp-endpoint.com/callback`), and
+3. set the name for your client.
+
+Run the script to register your client:
+```bash
+uv run registration.py
+```
+
+You should see something like
+
+```
+Client registered successfully!
+{
+  # detailed json response
+}
+
+=== Save these credentials ===
+Client ID: abc-123
+Client Secret: xyz-987
+```
+
+Take a note of the client id and client secret printed at the end, as you will need them in the next step.
+
+## 3. Configure your MCP
+Next, you need to configure your MCP to use the OAuth2 credentials you just created.
+In `main.py`, update these settings:
+
+```python
+auth_server = "<auth server url>"
+resource_server = "http://localhost:8000"  # This server's URL
+
+client_id = "<the client id returned by the registration.py script>"
+client_secret = "<the client secret returned by the registration.py script>"
+```
+
+## 4. Run the example
+
+With these in place, you can run the server using
+
+```python
+uv run main.py
+```
+
+This will start an MCP server protected by OAuth2.
+You can test it using an MCP client that supports OAuth2 authentication, such as [MCP Inspector](https://modelcontextprotocol.io/docs/tools/inspector).
+
+
+## Further reading
+More details on oauth authorization and the MCP protocol can be found at [https://modelcontextprotocol.io/specification/draft/basic/authorization](https://modelcontextprotocol.io/specification/draft/basic/authorization).

examples/oauth/protected_by_oauth/main.py

@@ -0,0 +1,88 @@
+"""
+Demonstration of an MCP agent server configured with OAuth.
+"""
+
+import asyncio
+from typing import Optional
+from pydantic import AnyHttpUrl
+
+from mcp_agent.core.context import Context as AppContext
+
+from mcp_agent.app import MCPApp
+from mcp_agent.server.app_server import create_mcp_server_for_app
+from mcp_agent.config import Settings, LoggerSettings, \
+    OAuthTokenStoreSettings, OAuthSettings, MCPAuthorizationServerSettings
+
+
+auth_server = "https://auth.mcp-agent.com" # the MCP Agent Cloud auth server, or replace with your own
+resource_server = "http://localhost:8000"  # This server's URL
+
+client_id = "<client id from registration.py>"
+client_secret = "<client secret from registration.py>"
+
+settings = Settings(
+        execution_engine="asyncio",
+        logger=LoggerSettings(level="info"),
+        authorization=MCPAuthorizationServerSettings(
+            enabled=True,
+            issuer_url=AnyHttpUrl(auth_server),
+            resource_server_url=AnyHttpUrl(resource_server),
+            client_id=client_id,
+            client_secret=client_secret,
+            required_scopes=["mcp"],
+            expected_audiences=[client_id],
+        ),
+        oauth=OAuthSettings(
+            callback_base_url=AnyHttpUrl(resource_server),
+            flow_timeout_seconds=300,
+            token_store=OAuthTokenStoreSettings(refresh_leeway_seconds=60),
+        )
+    )
+
+
+# Define the MCPApp instance. The server created for this app will advertise the
+# MCP logging capability and forward structured logs upstream to connected clients.
+app = MCPApp(
+    name="oauth_demo",
+    description="Basic agent server example",
+    settings=settings,
+)
+
+
+@app.tool(name="hello_world")
+async def hello(app_ctx: Optional[AppContext] = None) -> str:
+    # Use the context's app if available for proper logging with upstream_session
+    _app = app_ctx.app if app_ctx else app
+    # Ensure the app's logger is bound to the current context with upstream_session
+    if _app._logger and hasattr(_app._logger, "_bound_context"):
+        _app._logger._bound_context = app_ctx
+
+    if app_ctx.current_user:
+        user = app_ctx.current_user
+        if user.claims and "username" in user.claims:
+            return f"Hello, {user.claims['username']}!"
+        else:
+            return f"Hello, user with ID {user.subject}!"
+    else:
+        return "Hello, anonymous user!"
+
+async def main():
+    async with app.run() as agent_app:
+        # Log registered workflows and agent configurations
+        agent_app.logger.info(f"Creating MCP server for {agent_app.name}")
+
+        agent_app.logger.info("Registered workflows:")
+        for workflow_id in agent_app.workflows:
+            agent_app.logger.info(f"  - {workflow_id}")
+
+        # Create the MCP server that exposes both workflows and agent configurations,
+        # optionally using custom FastMCP settings
+        mcp_server = create_mcp_server_for_app(agent_app)
+        agent_app.logger.info(f"MCP Server settings: {mcp_server.settings}")
+
+        # Run the server
+        await mcp_server.run_sse_async()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/oauth/protected_by_oauth/registration.py

@@ -0,0 +1,60 @@
+import requests
+import json
+
+# Authorization server URL. This can either be the MCP Agent Clound authorization server (as currently configured),
+# or your own.
+auth_server_url = "https://auth.mcp-agent.com"
+redirect_uris = [
+    # These are the redirect URIs for MCP Inspector. Replace with your app's URIs.
+    "http://localhost:6274/oauth/callback",
+    "http://localhost:6274/oauth/callback/debug"
+]
+client_name = "My Python Application"
+
+# Fetch the registration endpoint dynamically from the .well-known/oauth-authorization-server details
+well_known_url = f"{auth_server_url}/.well-known/oauth-authorization-server"
+response = requests.get(well_known_url)
+
+if response.status_code == 200:
+    well_known_details = response.json()
+    registration_endpoint = well_known_details.get("registration_endpoint")
+    if not registration_endpoint:
+        raise ValueError("Registration endpoint not found in .well-known details")
+else:
+    raise ValueError(f"Failed to fetch .well-known details: {response.status_code}")
+
+
+# Client registration request
+registration_request = {
+    "client_name": client_name,
+    "redirect_uris": redirect_uris,
+    "grant_types": [
+        "authorization_code",
+        "refresh_token"
+    ],
+    "scope": "mcp",
+    # use client_secret_basic when testing with MCP Inspector
+    "token_endpoint_auth_method": "client_secret_basic"
+}
+
+print(f"Registering client at: {registration_endpoint}")
+
+# Register the client
+response = requests.post(
+    registration_endpoint,
+    json=registration_request,
+    headers={"Content-Type": "application/json"}
+)
+
+if response.status_code in [200, 201]:
+    client_info = response.json()
+    print("Client registered successfully!")
+    print(json.dumps(client_info, indent=2))
+
+    # Save credentials for later use
+    print("\n=== Save these credentials ===")
+    print(f"Client ID: {client_info['client_id']}")
+    print(f"Client Secret: {client_info['client_secret']}")
+else:
+    print(f"Registration failed with status {response.status_code}")
+    print(response.text)

schema/mcp-agent.config.schema.json

@@ -698,22 +698,7 @@
           "title": "Jwks Uri",
           "description": "Optional JWKS endpoint for validating JWT access tokens."
         },
-        "introspection_endpoint": {
-          "anyOf": [
-            {
-              "format": "uri",
-              "minLength": 1,
-              "type": "string"
-            },
-            {
-              "type": "null"
-            }
-          ],
-          "default": null,
-          "title": "Introspection Endpoint",
-          "description": "Optional OAuth introspection endpoint for opaque tokens."
-        },
-        "introspection_client_id": {
+        "client_id": {
           "anyOf": [
             {
               "type": "string"
@@ -723,10 +708,10 @@
             }
           ],
           "default": null,
-          "title": "Introspection Client Id",
+          "title": "Client Id",
           "description": "Client id to use when calling the introspection endpoint."
         },
-        "introspection_client_secret": {
+        "client_secret": {
           "anyOf": [
             {
               "type": "string"
@@ -736,7 +721,7 @@
             }
           ],
           "default": null,
-          "title": "Introspection Client Secret",
+          "title": "Client Secret",
           "description": "Client secret to use when calling the introspection endpoint."
         },
         "token_cache_ttl_seconds": {

src/mcp_agent/config.py

@@ -47,13 +47,10 @@ class MCPAuthorizationServerSettings(BaseModel):
     jwks_uri: AnyHttpUrl | None = None
     """Optional JWKS endpoint for validating JWT access tokens."""
 
-    introspection_endpoint: AnyHttpUrl | None = None
-    """Optional OAuth introspection endpoint for opaque tokens."""
-
-    introspection_client_id: str | None = None
+    client_id: str | None = None
     """Client id to use when calling the introspection endpoint."""
 
-    introspection_client_secret: str | None = None
+    client_secret: str | None = None
     """Client secret to use when calling the introspection endpoint."""
 
     token_cache_ttl_seconds: int = Field(300, ge=0)
@@ -1162,6 +1159,9 @@ def _find_config(cls, filenames: List[str]) -> Path | None:
         return None
 
 
+Settings.model_rebuild()
+
+
 class PreloadSettings(BaseSettings):
     """
     Class for preloaded settings of the MCP Agent application.

src/mcp_agent/oauth/metadata.py

@@ -32,6 +32,33 @@ async def fetch_authorization_server_metadata(
     return OAuthMetadata.model_validate(response.json())
 
 
+async def fetch_authorization_server_metadata_from_issuer(
+    client: httpx.AsyncClient,
+    issuer_url: str,
+) -> OAuthMetadata:
+    """Fetch OAuth authorization server metadata from the well-known endpoint.
+
+    Given an issuer URL, constructs the well-known OAuth authorization server
+    metadata URL and fetches the metadata.
+
+    Args:
+        client: HTTP client to use for the request
+        issuer_url: The issuer URL (e.g., "https://auth.example.com")
+
+    Returns:
+        OAuthMetadata containing authorization server metadata including introspection_endpoint
+    """
+    from httpx import URL
+
+    parsed_url = URL(issuer_url)
+    metadata_url = str(
+        parsed_url.copy_with(
+            path="/.well-known/oauth-authorization-server" + parsed_url.path
+        )
+    )
+    return await fetch_authorization_server_metadata(client, metadata_url)
+
+
 def select_authorization_server(
     metadata: ProtectedResourceMetadata,
     preferred: str | None = None,