Skip to main content
Glama

MCP Python SDK

by hoyoun86
MIT License

MCP Python SDK

Python implementation of the Model Context Protocol (MCP)

Table of Contents

Overview

The Model Context Protocol allows applications to provide context for LLMs in a standardized way, separating the concerns of providing context from the actual LLM interaction. This Python SDK implements the full MCP specification, making it easy to:

  • Build MCP clients that can connect to any MCP server
  • Create MCP servers that expose resources, prompts and tools
  • Use standard transports like stdio, SSE, and Streamable HTTP
  • Handle all MCP protocol messages and lifecycle events

Installation

Adding MCP to your python project

We recommend using uv to manage your Python projects.

If you haven't created a uv-managed project yet, create one:

uv init mcp-server-demo cd mcp-server-demo

Then add MCP to your project dependencies:

uv add "mcp[cli]"

Alternatively, for projects using pip for dependencies:

pip install "mcp[cli]"

Running the standalone MCP development tools

To run the mcp command with uv:

uv run mcp

Quickstart

Let's create a simple MCP server that exposes a calculator tool and some data:

# server.py from mcp.server.fastmcp import FastMCP # Create an MCP server mcp = FastMCP("Demo") # Add an addition tool @mcp.tool() def add(a: int, b: int) -> int: """Add two numbers""" return a + b # Add a dynamic greeting resource @mcp.resource("greeting://{name}") def get_greeting(name: str) -> str: """Get a personalized greeting""" return f"Hello, {name}!"

You can install this server in Claude Desktop and interact with it right away by running:

mcp install server.py

Alternatively, you can test it with the MCP Inspector:

mcp dev server.py

What is MCP?

The Model Context Protocol (MCP) lets you build servers that expose data and functionality to LLM applications in a secure, standardized way. Think of it like a web API, but specifically designed for LLM interactions. MCP servers can:

  • Expose data through Resources (think of these sort of like GET endpoints; they are used to load information into the LLM's context)
  • Provide functionality through Tools (sort of like POST endpoints; they are used to execute code or otherwise produce a side effect)
  • Define interaction patterns through Prompts (reusable templates for LLM interactions)
  • And more!

Core Concepts

Server

The FastMCP server is your core interface to the MCP protocol. It handles connection management, protocol compliance, and message routing:

# Add lifespan support for startup/shutdown with strong typing from contextlib import asynccontextmanager from collections.abc import AsyncIterator from dataclasses import dataclass from fake_database import Database # Replace with your actual DB type from mcp.server.fastmcp import FastMCP # Create a named server mcp = FastMCP("My App") # Specify dependencies for deployment and development mcp = FastMCP("My App", dependencies=["pandas", "numpy"]) @dataclass class AppContext: db: Database @asynccontextmanager async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]: """Manage application lifecycle with type-safe context""" # Initialize on startup db = await Database.connect() try: yield AppContext(db=db) finally: # Cleanup on shutdown await db.disconnect() # Pass lifespan to server mcp = FastMCP("My App", lifespan=app_lifespan) # Access type-safe lifespan context in tools @mcp.tool() def query_db() -> str: """Tool that uses initialized resources""" ctx = mcp.get_context() db = ctx.request_context.lifespan_context["db"] return db.query()

Resources

Resources are how you expose data to LLMs. They're similar to GET endpoints in a REST API - they provide data but shouldn't perform significant computation or have side effects:

from mcp.server.fastmcp import FastMCP mcp = FastMCP("My App") @mcp.resource("config://app", title="Application Configuration") def get_config() -> str: """Static configuration data""" return "App configuration here" @mcp.resource("users://{user_id}/profile", title="User Profile") def get_user_profile(user_id: str) -> str: """Dynamic user data""" return f"Profile data for user {user_id}"

Tools

Tools let LLMs take actions through your server. Unlike resources, tools are expected to perform computation and have side effects:

import httpx from mcp.server.fastmcp import FastMCP mcp = FastMCP("My App") @mcp.tool(title="BMI Calculator") def calculate_bmi(weight_kg: float, height_m: float) -> float: """Calculate BMI given weight in kg and height in meters""" return weight_kg / (height_m**2) @mcp.tool(title="Weather Fetcher") async def fetch_weather(city: str) -> str: """Fetch current weather for a city""" async with httpx.AsyncClient() as client: response = await client.get(f"https://api.weather.com/{city}") return response.text
Structured Output

Tools will return structured results by default, if their return type annotation is compatible. Otherwise, they will return unstructured results.

Structured output supports these return types:

  • Pydantic models (BaseModel subclasses)
  • TypedDicts
  • Dataclasses and other classes with type hints
  • dict[str, T] (where T is any JSON-serializable type)
  • Primitive types (str, int, float, bool, bytes, None) - wrapped in {"result": value}
  • Generic types (list, tuple, Union, Optional, etc.) - wrapped in {"result": value}

Classes without type hints cannot be serialized for structured output. Only classes with properly annotated attributes will be converted to Pydantic models for schema generation and validation.

Structured results are automatically validated against the output schema generated from the annotation. This ensures the tool returns well-typed, validated data that clients can easily process.

Note: For backward compatibility, unstructured results are also returned. Unstructured results are provided for backward compatibility with previous versions of the MCP specification, and are quirks-compatible with previous versions of FastMCP in the current version of the SDK.

Note: In cases where a tool function's return type annotation causes the tool to be classified as structured and this is undesirable, the classification can be suppressed by passing structured_output=False to the @tool decorator.

from mcp.server.fastmcp import FastMCP from pydantic import BaseModel, Field from typing import TypedDict mcp = FastMCP("Weather Service") # Using Pydantic models for rich structured data class WeatherData(BaseModel): temperature: float = Field(description="Temperature in Celsius") humidity: float = Field(description="Humidity percentage") condition: str wind_speed: float @mcp.tool() def get_weather(city: str) -> WeatherData: """Get structured weather data""" return WeatherData( temperature=22.5, humidity=65.0, condition="partly cloudy", wind_speed=12.3 ) # Using TypedDict for simpler structures class LocationInfo(TypedDict): latitude: float longitude: float name: str @mcp.tool() def get_location(address: str) -> LocationInfo: """Get location coordinates""" return LocationInfo(latitude=51.5074, longitude=-0.1278, name="London, UK") # Using dict[str, Any] for flexible schemas @mcp.tool() def get_statistics(data_type: str) -> dict[str, float]: """Get various statistics""" return {"mean": 42.5, "median": 40.0, "std_dev": 5.2} # Ordinary classes with type hints work for structured output class UserProfile: name: str age: int email: str | None = None def __init__(self, name: str, age: int, email: str | None = None): self.name = name self.age = age self.email = email @mcp.tool() def get_user(user_id: str) -> UserProfile: """Get user profile - returns structured data""" return UserProfile(name="Alice", age=30, email="alice@example.com") # Classes WITHOUT type hints cannot be used for structured output class UntypedConfig: def __init__(self, setting1, setting2): self.setting1 = setting1 self.setting2 = setting2 @mcp.tool() def get_config() -> UntypedConfig: """This returns unstructured output - no schema generated""" return UntypedConfig("value1", "value2") # Lists and other types are wrapped automatically @mcp.tool() def list_cities() -> list[str]: """Get a list of cities""" return ["London", "Paris", "Tokyo"] # Returns: {"result": ["London", "Paris", "Tokyo"]} @mcp.tool() def get_temperature(city: str) -> float: """Get temperature as a simple float""" return 22.5 # Returns: {"result": 22.5}

Prompts

Prompts are reusable templates that help LLMs interact with your server effectively:

from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp.prompts import base mcp = FastMCP("My App") @mcp.prompt(title="Code Review") def review_code(code: str) -> str: return f"Please review this code:\n\n{code}" @mcp.prompt(title="Debug Assistant") def debug_error(error: str) -> list[base.Message]: return [ base.UserMessage("I'm seeing this error:"), base.UserMessage(error), base.AssistantMessage("I'll help debug that. What have you tried so far?"), ]

Images

FastMCP provides an Image class that automatically handles image data:

from mcp.server.fastmcp import FastMCP, Image from PIL import Image as PILImage mcp = FastMCP("My App") @mcp.tool() def create_thumbnail(image_path: str) -> Image: """Create a thumbnail from an image""" img = PILImage.open(image_path) img.thumbnail((100, 100)) return Image(data=img.tobytes(), format="png")

Context

The Context object gives your tools and resources access to MCP capabilities:

from mcp.server.fastmcp import FastMCP, Context mcp = FastMCP("My App") @mcp.tool() async def long_task(files: list[str], ctx: Context) -> str: """Process multiple files with progress tracking""" for i, file in enumerate(files): ctx.info(f"Processing {file}") await ctx.report_progress(i, len(files)) data, mime_type = await ctx.read_resource(f"file://{file}") return "Processing complete"

Completions

MCP supports providing completion suggestions for prompt arguments and resource template parameters. With the context parameter, servers can provide completions based on previously resolved values:

Client usage:

from mcp.client.session import ClientSession from mcp.types import ResourceTemplateReference async def use_completion(session: ClientSession): # Complete without context result = await session.complete( ref=ResourceTemplateReference( type="ref/resource", uri="github://repos/{owner}/{repo}" ), argument={"name": "owner", "value": "model"}, ) # Complete with context - repo suggestions based on owner result = await session.complete( ref=ResourceTemplateReference( type="ref/resource", uri="github://repos/{owner}/{repo}" ), argument={"name": "repo", "value": "test"}, context_arguments={"owner": "modelcontextprotocol"}, )

Server implementation:

from mcp.server import Server from mcp.types import ( Completion, CompletionArgument, CompletionContext, PromptReference, ResourceTemplateReference, ) server = Server("example-server") @server.completion() async def handle_completion( ref: PromptReference | ResourceTemplateReference, argument: CompletionArgument, context: CompletionContext | None, ) -> Completion | None: if isinstance(ref, ResourceTemplateReference): if ref.uri == "github://repos/{owner}/{repo}" and argument.name == "repo": # Use context to provide owner-specific repos if context and context.arguments: owner = context.arguments.get("owner") if owner == "modelcontextprotocol": repos = ["python-sdk", "typescript-sdk", "specification"] # Filter based on partial input filtered = [r for r in repos if r.startswith(argument.value)] return Completion(values=filtered) return None

Elicitation

Request additional information from users during tool execution:

from mcp.server.fastmcp import FastMCP, Context from mcp.server.elicitation import ( AcceptedElicitation, DeclinedElicitation, CancelledElicitation, ) from pydantic import BaseModel, Field mcp = FastMCP("Booking System") @mcp.tool() async def book_table(date: str, party_size: int, ctx: Context) -> str: """Book a table with confirmation""" # Schema must only contain primitive types (str, int, float, bool) class ConfirmBooking(BaseModel): confirm: bool = Field(description="Confirm booking?") notes: str = Field(default="", description="Special requests") result = await ctx.elicit( message=f"Confirm booking for {party_size} on {date}?", schema=ConfirmBooking ) match result: case AcceptedElicitation(data=data): if data.confirm: return f"Booked! Notes: {data.notes or 'None'}" return "Booking cancelled" case DeclinedElicitation(): return "Booking declined" case CancelledElicitation(): return "Booking cancelled"

The elicit() method returns an ElicitationResult with:

  • action: "accept", "decline", or "cancel"
  • data: The validated response (only when accepted)
  • validation_error: Any validation error message

Authentication

Authentication can be used by servers that want to expose tools accessing protected resources.

mcp.server.auth implements OAuth 2.1 resource server functionality, where MCP servers act as Resource Servers (RS) that validate tokens issued by separate Authorization Servers (AS). This follows the MCP authorization specification and implements RFC 9728 (Protected Resource Metadata) for AS discovery.

MCP servers can use authentication by providing an implementation of the TokenVerifier protocol:

from mcp import FastMCP from mcp.server.auth.provider import TokenVerifier, TokenInfo from mcp.server.auth.settings import AuthSettings class MyTokenVerifier(TokenVerifier): # Implement token validation logic (typically via token introspection) async def verify_token(self, token: str) -> TokenInfo: # Verify with your authorization server ... mcp = FastMCP( "My App", token_verifier=MyTokenVerifier(), auth=AuthSettings( issuer_url="https://auth.example.com", resource_server_url="http://localhost:3001", required_scopes=["mcp:read", "mcp:write"], ), )

For a complete example with separate Authorization Server and Resource Server implementations, see examples/servers/simple-auth/.

Architecture:

  • Authorization Server (AS): Handles OAuth flows, user authentication, and token issuance
  • Resource Server (RS): Your MCP server that validates tokens and serves protected resources
  • Client: Discovers AS through RFC 9728, obtains tokens, and uses them with the MCP server

See TokenVerifier for more details on implementing token validation.

Running Your Server

Development Mode

The fastest way to test and debug your server is with the MCP Inspector:

mcp dev server.py # Add dependencies mcp dev server.py --with pandas --with numpy # Mount local code mcp dev server.py --with-editable .

Claude Desktop Integration

Once your server is ready, install it in Claude Desktop:

mcp install server.py # Custom name mcp install server.py --name "My Analytics Server" # Environment variables mcp install server.py -v API_KEY=abc123 -v DB_URL=postgres://... mcp install server.py -f .env

Direct Execution

For advanced scenarios like custom deployments:

from mcp.server.fastmcp import FastMCP mcp = FastMCP("My App") if __name__ == "__main__": mcp.run()

Run it with:

python server.py # or mcp run server.py

Note that mcp run or mcp dev only supports server using FastMCP and not the low-level server variant.

Streamable HTTP Transport

Note: Streamable HTTP transport is superseding SSE transport for production deployments.

from mcp.server.fastmcp import FastMCP # Stateful server (maintains session state) mcp = FastMCP("StatefulServer") # Stateless server (no session persistence) mcp = FastMCP("StatelessServer", stateless_http=True) # Stateless server (no session persistence, no sse stream with supported client) mcp = FastMCP("StatelessServer", stateless_http=True, json_response=True) # Run server with streamable_http transport mcp.run(transport="streamable-http")

You can mount multiple FastMCP servers in a FastAPI application:

# echo.py from mcp.server.fastmcp import FastMCP mcp = FastMCP(name="EchoServer", stateless_http=True) @mcp.tool(description="A simple echo tool") def echo(message: str) -> str: return f"Echo: {message}"
# math.py from mcp.server.fastmcp import FastMCP mcp = FastMCP(name="MathServer", stateless_http=True) @mcp.tool(description="A simple add tool") def add_two(n: int) -> int: return n + 2
# main.py import contextlib from fastapi import FastAPI from mcp.echo import echo from mcp.math import math # Create a combined lifespan to manage both session managers @contextlib.asynccontextmanager async def lifespan(app: FastAPI): async with contextlib.AsyncExitStack() as stack: await stack.enter_async_context(echo.mcp.session_manager.run()) await stack.enter_async_context(math.mcp.session_manager.run()) yield app = FastAPI(lifespan=lifespan) app.mount("/echo", echo.mcp.streamable_http_app()) app.mount("/math", math.mcp.streamable_http_app())

For low level server with Streamable HTTP implementations, see:

The streamable HTTP transport supports:

  • Stateful and stateless operation modes
  • Resumability with event stores
  • JSON or SSE response formats
  • Better scalability for multi-node deployments

Mounting to an Existing ASGI Server

Note: SSE transport is being superseded by Streamable HTTP transport.

By default, SSE servers are mounted at /sse and Streamable HTTP servers are mounted at /mcp. You can customize these paths using the methods described below.

You can mount the SSE server to an existing ASGI server using the sse_app method. This allows you to integrate the SSE server with other ASGI applications.

from starlette.applications import Starlette from starlette.routing import Mount, Host from mcp.server.fastmcp import FastMCP mcp = FastMCP("My App") # Mount the SSE server to the existing ASGI server app = Starlette( routes=[ Mount('/', app=mcp.sse_app()), ] ) # or dynamically mount as host app.router.routes.append(Host('mcp.acme.corp', app=mcp.sse_app()))

When mounting multiple MCP servers under different paths, you can configure the mount path in several ways:

from starlette.applications import Starlette from starlette.routing import Mount from mcp.server.fastmcp import FastMCP # Create multiple MCP servers github_mcp = FastMCP("GitHub API") browser_mcp = FastMCP("Browser") curl_mcp = FastMCP("Curl") search_mcp = FastMCP("Search") # Method 1: Configure mount paths via settings (recommended for persistent configuration) github_mcp.settings.mount_path = "/github" browser_mcp.settings.mount_path = "/browser" # Method 2: Pass mount path directly to sse_app (preferred for ad-hoc mounting) # This approach doesn't modify the server's settings permanently # Create Starlette app with multiple mounted servers app = Starlette( routes=[ # Using settings-based configuration Mount("/github", app=github_mcp.sse_app()), Mount("/browser", app=browser_mcp.sse_app()), # Using direct mount path parameter Mount("/curl", app=curl_mcp.sse_app("/curl")), Mount("/search", app=search_mcp.sse_app("/search")), ] ) # Method 3: For direct execution, you can also pass the mount path to run() if __name__ == "__main__": search_mcp.run(transport="sse", mount_path="/search")

For more information on mounting applications in Starlette, see the Starlette documentation.

Examples

Echo Server

A simple server demonstrating resources, tools, and prompts:

from mcp.server.fastmcp import FastMCP mcp = FastMCP("Echo") @mcp.resource("echo://{message}") def echo_resource(message: str) -> str: """Echo a message as a resource""" return f"Resource echo: {message}" @mcp.tool() def echo_tool(message: str) -> str: """Echo a message as a tool""" return f"Tool echo: {message}" @mcp.prompt() def echo_prompt(message: str) -> str: """Create an echo prompt""" return f"Please process this message: {message}"

SQLite Explorer

A more complex example showing database integration:

import sqlite3 from mcp.server.fastmcp import FastMCP mcp = FastMCP("SQLite Explorer") @mcp.resource("schema://main") def get_schema() -> str: """Provide the database schema as a resource""" conn = sqlite3.connect("database.db") schema = conn.execute("SELECT sql FROM sqlite_master WHERE type='table'").fetchall() return "\n".join(sql[0] for sql in schema if sql[0]) @mcp.tool() def query_data(sql: str) -> str: """Execute SQL queries safely""" conn = sqlite3.connect("database.db") try: result = conn.execute(sql).fetchall() return "\n".join(str(row) for row in result) except Exception as e: return f"Error: {str(e)}"

Advanced Usage

Low-Level Server

For more control, you can use the low-level server implementation directly. This gives you full access to the protocol and allows you to customize every aspect of your server, including lifecycle management through the lifespan API:

from contextlib import asynccontextmanager from collections.abc import AsyncIterator from fake_database import Database # Replace with your actual DB type from mcp.server import Server @asynccontextmanager async def server_lifespan(server: Server) -> AsyncIterator[dict]: """Manage server startup and shutdown lifecycle.""" # Initialize resources on startup db = await Database.connect() try: yield {"db": db} finally: # Clean up on shutdown await db.disconnect() # Pass lifespan to server server = Server("example-server", lifespan=server_lifespan) # Access lifespan context in handlers @server.call_tool() async def query_db(name: str, arguments: dict) -> list: ctx = server.request_context db = ctx.lifespan_context["db"] return await db.query(arguments["query"])

The lifespan API provides:

  • A way to initialize resources when the server starts and clean them up when it stops
  • Access to initialized resources through the request context in handlers
  • Type-safe context passing between lifespan and request handlers
import mcp.server.stdio import mcp.types as types from mcp.server.lowlevel import NotificationOptions, Server from mcp.server.models import InitializationOptions # Create a server instance server = Server("example-server") @server.list_prompts() async def handle_list_prompts() -> list[types.Prompt]: return [ types.Prompt( name="example-prompt", description="An example prompt template", arguments=[ types.PromptArgument( name="arg1", description="Example argument", required=True ) ], ) ] @server.get_prompt() async def handle_get_prompt( name: str, arguments: dict[str, str] | None ) -> types.GetPromptResult: if name != "example-prompt": raise ValueError(f"Unknown prompt: {name}") return types.GetPromptResult( description="Example prompt", messages=[ types.PromptMessage( role="user", content=types.TextContent(type="text", text="Example prompt text"), ) ], ) async def run(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_name="example", server_version="0.1.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), ) if __name__ == "__main__": import asyncio asyncio.run(run())

Caution: The mcp run and mcp dev tool doesn't support low-level server.

Structured Output Support

The low-level server supports structured output for tools, allowing you to return both human-readable content and machine-readable structured data. Tools can define an outputSchema to validate their structured output:

from types import Any import mcp.types as types from mcp.server.lowlevel import Server server = Server("example-server") @server.list_tools() async def list_tools() -> list[types.Tool]: return [ types.Tool( name="calculate", description="Perform mathematical calculations", inputSchema={ "type": "object", "properties": { "expression": {"type": "string", "description": "Math expression"} }, "required": ["expression"], }, outputSchema={ "type": "object", "properties": { "result": {"type": "number"}, "expression": {"type": "string"}, }, "required": ["result", "expression"], }, ) ] @server.call_tool() async def call_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]: if name == "calculate": expression = arguments["expression"] try: result = eval(expression) # Use a safe math parser structured = {"result": result, "expression": expression} # low-level server will validate structured output against the tool's # output schema, and automatically serialize it into a TextContent block # for backwards compatibility with pre-2025-06-18 clients. return structured except Exception as e: raise ValueError(f"Calculation error: {str(e)}")

Tools can return data in three ways:

  1. Content only: Return a list of content blocks (default behavior before spec revision 2025-06-18)
  2. Structured data only: Return a dictionary that will be serialized to JSON (Introduced in spec revision 2025-06-18)
  3. Both: Return a tuple of (content, structured_data) preferred option to use for backwards compatibility

When an outputSchema is defined, the server automatically validates the structured output against the schema. This ensures type safety and helps catch errors early.

Writing MCP Clients

The SDK provides a high-level client interface for connecting to MCP servers using various transports:

from mcp import ClientSession, StdioServerParameters, types from mcp.client.stdio import stdio_client # Create server parameters for stdio connection server_params = StdioServerParameters( command="python", # Executable args=["example_server.py"], # Optional command line arguments env=None, # Optional environment variables ) # Optional: create a sampling callback async def handle_sampling_message( message: types.CreateMessageRequestParams, ) -> types.CreateMessageResult: return types.CreateMessageResult( role="assistant", content=types.TextContent( type="text", text="Hello, world! from model", ), model="gpt-3.5-turbo", stopReason="endTurn", ) async def run(): async with stdio_client(server_params) as (read, write): async with ClientSession( read, write, sampling_callback=handle_sampling_message ) as session: # Initialize the connection await session.initialize() # List available prompts prompts = await session.list_prompts() # Get a prompt prompt = await session.get_prompt( "example-prompt", arguments={"arg1": "value"} ) # List available resources resources = await session.list_resources() # List available tools tools = await session.list_tools() # Read a resource content, mime_type = await session.read_resource("file://some/path") # Call a tool result = await session.call_tool("tool-name", arguments={"arg1": "value"}) if __name__ == "__main__": import asyncio asyncio.run(run())

Clients can also connect using Streamable HTTP transport:

from mcp.client.streamable_http import streamablehttp_client from mcp import ClientSession async def main(): # Connect to a streamable HTTP server async with streamablehttp_client("example/mcp") as ( read_stream, write_stream, _, ): # Create a session using the client streams async with ClientSession(read_stream, write_stream) as session: # Initialize the connection await session.initialize() # Call a tool tool_result = await session.call_tool("echo", {"message": "hello"})

Client Display Utilities

When building MCP clients, the SDK provides utilities to help display human-readable names for tools, resources, and prompts:

from mcp.shared.metadata_utils import get_display_name from mcp.client.session import ClientSession async def display_tools(session: ClientSession): """Display available tools with human-readable names""" tools_response = await session.list_tools() for tool in tools_response.tools: # get_display_name() returns the title if available, otherwise the name display_name = get_display_name(tool) print(f"Tool: {display_name}") if tool.description: print(f" {tool.description}") async def display_resources(session: ClientSession): """Display available resources with human-readable names""" resources_response = await session.list_resources() for resource in resources_response.resources: display_name = get_display_name(resource) print(f"Resource: {display_name} ({resource.uri})")

The get_display_name() function implements the proper precedence rules for displaying names:

  • For tools: title > annotations.title > name
  • For other objects: title > name

This ensures your client UI shows the most user-friendly names that servers provide.

OAuth Authentication for Clients

The SDK includes authorization support for connecting to protected MCP servers:

from mcp.client.auth import OAuthClientProvider, TokenStorage from mcp.client.session import ClientSession from mcp.client.streamable_http import streamablehttp_client from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken class CustomTokenStorage(TokenStorage): """Simple in-memory token storage implementation.""" async def get_tokens(self) -> OAuthToken | None: pass async def set_tokens(self, tokens: OAuthToken) -> None: pass async def get_client_info(self) -> OAuthClientInformationFull | None: pass async def set_client_info(self, client_info: OAuthClientInformationFull) -> None: pass async def main(): # Set up OAuth authentication oauth_auth = OAuthClientProvider( server_url="https://api.example.com", client_metadata=OAuthClientMetadata( client_name="My Client", redirect_uris=["http://localhost:3000/callback"], grant_types=["authorization_code", "refresh_token"], response_types=["code"], ), storage=CustomTokenStorage(), redirect_handler=lambda url: print(f"Visit: {url}"), callback_handler=lambda: ("auth_code", None), ) # Use with streamable HTTP client async with streamablehttp_client( "https://api.example.com/mcp", auth=oauth_auth ) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() # Authenticated session ready

For a complete working example, see examples/clients/simple-auth-client/.

MCP Primitives

The MCP protocol defines three core primitives that servers can implement:

PrimitiveControlDescriptionExample Use
PromptsUser-controlledInteractive templates invoked by user choiceSlash commands, menu options
ResourcesApplication-controlledContextual data managed by the client applicationFile contents, API responses
ToolsModel-controlledFunctions exposed to the LLM to take actionsAPI calls, data updates

Server Capabilities

MCP servers declare capabilities during initialization:

CapabilityFeature FlagDescription
promptslistChangedPrompt template management
resourcessubscribelistChangedResource exposure and updates
toolslistChangedTool discovery and execution
logging-Server logging configuration
completion-Argument completion suggestions

Documentation

Contributing

We are passionate about supporting contributors of all levels of experience and would love to see you get involved in the project. See the contributing guide to get started.

License

This project is licensed under the MIT License - see the LICENSE file for details.

Related MCP Servers

  • -
    security
    A
    license
    -
    quality
    A Model Context Protocol server that extracts and analyzes Python code structures, focusing on import/export relationships between files to help LLMs understand code context.
    Last updated -
    4
    Python
    MIT License
  • -
    security
    A
    license
    -
    quality
    A streamlined foundation for building Model Context Protocol servers in Python, designed to make AI-assisted development of MCP tools easier and more efficient.
    Last updated -
    14
    Python
    MIT License
  • -
    security
    F
    license
    -
    quality
    A Python-based implementation of the Model Context Protocol that enables communication between a model context management server and client through a request-response architecture.
    Last updated -
    Python
    • Linux
    • Apple
  • A
    security
    F
    license
    A
    quality
    A Model Context Protocol server that allows LLMs to interact with Python environments, execute code, and manage files within a specified working directory.
    Last updated -
    9
    42
    Python
    • Linux
    • Apple

View all related MCP servers

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/hoyoun86/mcp'

If you have feedback or need assistance with the MCP directory API, please join our Discord server