tessl/pypi-mcp

Model Context Protocol SDK for building MCP servers and clients in Python

—

Pending

Overview

Eval results

Files

FastMCP Server Framework

Name: tessl/pypi-mcp
Author: tessl

High-level server framework using decorators for rapid MCP server development. FastMCP provides a modern, intuitive API for building MCP servers with built-in support for tools, resources, prompts, HTTP endpoints, authentication, and multiple transport protocols.

Capabilities

FastMCP Server Class

Main server class with decorator-based configuration for tools, resources, prompts, and HTTP endpoints.

class FastMCP:
    def __init__(
        self,
        name: str | None = None,
        instructions: str | None = None,
        auth_server_provider: OAuthAuthorizationServerProvider | None = None,
        token_verifier: TokenVerifier | None = None,
        *,
        tools: list[Tool] | None = None,
        debug: bool = False,
        log_level: str = "INFO",
        host: str = "127.0.0.1",
        port: int = 8000,
        **kwargs
    ):
        """
        Initialize FastMCP server instance.

        Parameters:
        - name: Server name for identification
        - instructions: Server instructions/description
        - auth_server_provider: OAuth authorization server provider
        - token_verifier: Token verification handler
        - tools: Pre-defined tools list
        - debug: Enable debug mode
        - log_level: Logging level (DEBUG, INFO, WARNING, ERROR)
        - host: Server host address
        - port: Server port number
        - **kwargs: Additional configuration options
        """

    def tool(
        self,
        name: str | None = None,
        description: str | None = None,
        **kwargs
    ) -> Callable:
        """
        Decorator for registering tool functions.

        Parameters:
        - name: Tool name (defaults to function name)
        - description: Tool description (defaults to docstring)
        - **kwargs: Additional tool metadata

        Returns:
        Decorator function for tool registration
        """

    def resource(
        self,
        uri: str | None = None,
        name: str | None = None,
        description: str | None = None,
        mime_type: str | None = None,
        **kwargs
    ) -> Callable:
        """
        Decorator for registering resource handlers.

        Parameters:
        - uri: Resource URI pattern
        - name: Resource name (defaults to function name)
        - description: Resource description (defaults to docstring)
        - mime_type: Resource MIME type
        - **kwargs: Additional resource metadata

        Returns:
        Decorator function for resource registration
        """

    def prompt(
        self,
        name: str | None = None,
        description: str | None = None,
        **kwargs
    ) -> Callable:
        """
        Decorator for registering prompt templates.

        Parameters:
        - name: Prompt name (defaults to function name)
        - description: Prompt description (defaults to docstring)
        - **kwargs: Additional prompt metadata

        Returns:
        Decorator function for prompt registration
        """

    def get(self, path: str, **kwargs) -> Callable:
        """
        Decorator for HTTP GET endpoint handlers.

        Parameters:
        - path: URL path pattern
        - **kwargs: Additional endpoint options

        Returns:
        Decorator function for endpoint registration
        """

    def post(self, path: str, **kwargs) -> Callable:
        """
        Decorator for HTTP POST endpoint handlers.

        Parameters:
        - path: URL path pattern
        - **kwargs: Additional endpoint options

        Returns:
        Decorator function for endpoint registration
        """

    def put(self, path: str, **kwargs) -> Callable:
        """
        Decorator for HTTP PUT endpoint handlers.

        Parameters:
        - path: URL path pattern
        - **kwargs: Additional endpoint options

        Returns:
        Decorator function for endpoint registration
        """

    def delete(self, path: str, **kwargs) -> Callable:
        """
        Decorator for HTTP DELETE endpoint handlers.

        Parameters:
        - path: URL path pattern
        - **kwargs: Additional endpoint options

        Returns:
        Decorator function for endpoint registration
        """

    async def run_stdio_async(self) -> None:
        """
        Run the server using stdio transport asynchronously.
        """

    async def run_sse_async(
        self,
        mount_path: str | None = None,
        **kwargs
    ) -> None:
        """
        Run the server using Server-Sent Events transport asynchronously.

        Parameters:
        - mount_path: Optional mount path for SSE endpoint
        - **kwargs: Additional server options
        """

    async def run_streamable_http_async(self) -> None:
        """
        Run the server using streamable HTTP transport asynchronously.
        """

    def run(
        self,
        transport: Literal["stdio", "sse", "streamable-http"] = "stdio",
        mount_path: str | None = None,
    ) -> None:
        """
        Run the FastMCP server synchronously.

        Parameters:
        - transport: Transport protocol ("stdio", "sse", or "streamable-http")
        - mount_path: Optional mount path for SSE transport
        """

Context Access

Request context object providing access to session information and request metadata within handler functions.

class Context:
    @property
    def request_id(self) -> str:
        """Current request identifier."""

    @property
    def client_session(self) -> ServerSession:
        """Current client session."""

    @property
    def user_id(self) -> str | None:
        """Authenticated user ID (if authentication enabled)."""

    @property
    def request_metadata(self) -> dict[str, Any]:
        """Request metadata and headers."""

    async def send_progress(
        self,
        progress: float,
        total: float | None = None,
        message: str | None = None
    ) -> None:
        """
        Send progress notification to client.

        Parameters:
        - progress: Current progress value
        - total: Total expected value
        - message: Progress message
        """

    async def send_log(
        self,
        level: LoggingLevel,
        message: str,
        **kwargs
    ) -> None:
        """
        Send log message to client.

        Parameters:
        - level: Log level
        - message: Log message
        - **kwargs: Additional log data
        """

Utility Types

Additional types and utilities for FastMCP development.

class Image:
    def __init__(
        self,
        data: bytes,
        mime_type: str = "image/png",
        **kwargs
    ):
        """
        Image data container for FastMCP.

        Parameters:
        - data: Image data bytes
        - mime_type: Image MIME type
        - **kwargs: Additional metadata
        """

    @classmethod
    def from_file(cls, path: str) -> "Image":
        """
        Load image from file path.

        Parameters:
        - path: File path to image

        Returns:
        Image instance
        """

    @classmethod
    def from_url(cls, url: str) -> "Image":
        """
        Load image from URL.

        Parameters:
        - url: Image URL

        Returns:
        Image instance
        """

    def to_base64(self) -> str:
        """
        Convert image to base64 string.

        Returns:
        Base64 encoded image data
        """

Usage Examples

Basic FastMCP Server

from mcp.server import FastMCP
import asyncio

# Create server instance
app = FastMCP("example-server", instructions="An example MCP server")

@app.tool()
async def calculate(operation: str, a: float, b: float) -> float:
    """Perform basic mathematical operations."""
    if operation == "add":
        return a + b
    elif operation == "subtract":
        return a - b
    elif operation == "multiply":
        return a * b
    elif operation == "divide":
        if b == 0:
            raise ValueError("Cannot divide by zero")
        return a / b
    else:
        raise ValueError(f"Unknown operation: {operation}")

@app.resource("config://settings")
async def get_settings() -> str:
    """Get application configuration."""
    return "debug=true\nlog_level=INFO\nmax_connections=100"

@app.prompt()
async def code_review_prompt(language: str, code: str) -> str:
    """Generate code review prompt."""
    return f"""Please review this {language} code:

```{language}
{code}

Focus on:

Code quality and best practices
Potential bugs or issues
Performance considerations
Security concerns """

Run the server

if name == "main": app.run("stdio")

### Server with HTTP Endpoints

```python
from mcp.server import FastMCP, Context
from fastapi import HTTPException

app = FastMCP("api-server")

@app.get("/health")
async def health_check():
    """Health check endpoint."""
    return {"status": "healthy", "timestamp": "2024-01-01T00:00:00Z"}

@app.post("/data")
async def process_data(data: dict):
    """Process submitted data."""
    if not data:
        raise HTTPException(status_code=400, detail="No data provided")
    
    # Process the data
    result = {"processed": True, "items": len(data)}
    return result

@app.tool()
async def get_api_status(ctx: Context) -> dict:
    """Get current API server status."""
    return {
        "request_id": ctx.request_id,
        "server": "api-server",
        "status": "running"
    }

# Run with HTTP transport
if __name__ == "__main__":
    app.run("sse")

Server with Authentication

from mcp.server import FastMCP, Context
from mcp.server.auth import AuthSettings, ProviderTokenVerifier

# Configure authentication
auth_settings = AuthSettings(
    client_id="your-client-id",
    client_secret="your-client-secret",  
    authorization_endpoint="https://auth.example.com/oauth/authorize",
    token_endpoint="https://auth.example.com/oauth/token"
)

token_verifier = ProviderTokenVerifier(auth_settings)

app = FastMCP(
    "secure-server",
    auth_server_provider=auth_settings,
    token_verifier=token_verifier
)

@app.tool()
async def get_user_data(ctx: Context) -> dict:
    """Get data for authenticated user."""
    user_id = ctx.user_id
    if not user_id:
        raise ValueError("Authentication required")
    
    return {
        "user_id": user_id,
        "data": f"User data for {user_id}"
    }

@app.resource("user://profile")
async def user_profile(ctx: Context) -> str:
    """Get user profile information."""
    user_id = ctx.user_id
    if not user_id:
        raise ValueError("Authentication required")
        
    return f"Profile data for user: {user_id}"

if __name__ == "__main__":
    app.run("sse")

Progress Reporting

from mcp.server import FastMCP, Context
import asyncio

app = FastMCP("progress-server")

@app.tool()
async def long_running_task(ctx: Context, iterations: int = 100) -> str:
    """Demonstrate progress reporting during long operations."""
    for i in range(iterations):
        # Simulate work
        await asyncio.sleep(0.1)
        
        # Report progress
        await ctx.send_progress(
            progress=i + 1,
            total=iterations,
            message=f"Processing item {i + 1}/{iterations}"
        )
    
    return f"Completed {iterations} iterations"

@app.tool()  
async def file_processor(ctx: Context, files: list[str]) -> dict:
    """Process multiple files with progress tracking."""
    results = {}
    
    for idx, filename in enumerate(files):
        await ctx.send_progress(
            progress=idx,
            total=len(files), 
            message=f"Processing {filename}"
        )
        
        # Simulate file processing
        await asyncio.sleep(0.5)
        results[filename] = f"Processed {filename}"
    
    return {"processed_files": results, "total": len(files)}

if __name__ == "__main__":
    app.run("stdio")

Resource Templates and Dynamic URIs

from mcp.server import FastMCP
import os

app = FastMCP("file-server")

@app.resource("file://{path}")
async def read_file(path: str) -> str:
    """Read file content by path."""
    if not os.path.exists(path):
        raise FileNotFoundError(f"File not found: {path}")
    
    with open(path, 'r') as f:
        return f.read()

@app.resource("config://{section}/{key}")
async def get_config_value(section: str, key: str) -> str:
    """Get configuration value by section and key."""
    config = {
        "database": {"host": "localhost", "port": "5432"},
        "api": {"key": "secret", "timeout": "30"}
    }
    
    if section not in config:
        raise ValueError(f"Unknown section: {section}")
    
    if key not in config[section]:
        raise ValueError(f"Unknown key: {key}")
    
    return config[section][key]

@app.tool()
async def list_files(directory: str = ".") -> list[str]:
    """List files in a directory."""
    if not os.path.isdir(directory):
        raise ValueError(f"Not a directory: {directory}")
    
    return [f for f in os.listdir(directory) if os.path.isfile(os.path.join(directory, f))]

if __name__ == "__main__":
    app.run("stdio")

Install with Tessl CLI