Version 1.5.0

from mcp_use.server import MCPServer

server = MCPServer(
    name="my-awesome-server",
    version="1.0.0",
    instructions="A production-ready MCP server",
    debug=True,  # Enable development tools
    pretty_print_jsonrpc=True,  # Beautiful JSON-RPC logs
)

@server.tool()
def calculate(expression: str) -> float:
    """Evaluate a mathematical expression."""
    return eval(expression)

# Run with multiple transport options
server.run(
    transport="streamable-http",  # or "stdio", "sse"
    host="0.0.0.0",
    port=8000,
    reload=True  # Auto-reload on code changes
)

from mcp_use.server import MCPServer, MCPRouter

# routes/math.py
math_router = MCPRouter()

@math_router.tool()
def add(a: float, b: float) -> float:
    """Add two numbers."""
    return a + b

@math_router.tool()
def multiply(a: float, b: float) -> float:
    """Multiply two numbers."""
    return a * b

# routes/weather.py
weather_router = MCPRouter()

@weather_router.resource(uri="weather://current/{city}")
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"Weather data for {city}"

# main.py
server = MCPServer(name="multi-module-server")
server.include_router(math_router, prefix="math")  # Tools: math_add, math_multiply
server.include_router(weather_router, prefix="weather")
server.include_router(admin_router, enabled=False)  # Conditionally disable routers

from mcp_use.server import MCPServer, Context

server = MCPServer(name="context-demo")

@server.tool()
async def smart_search(query: str, ctx: Context) -> str:
    """Intelligent search with LLM refinement."""
    # Sampling: Request LLM assistance from the client
    result = await ctx.sample(
        messages=f"Refine this search query: {query}",
        max_tokens=100,
        temperature=0.7,
        model_preferences={"hints": [{"name": "claude-3-5-sonnet"}]}
    )
    refined_query = result.content.text
    
    # Elicitation: Get structured input from user
    from dataclasses import dataclass
    
    @dataclass
    class SearchFilters:
        date_range: str
        categories: list[str]
        max_results: int = 10
    
    filters = await ctx.elicit(
        message="Please specify search filters",
        schema=SearchFilters  # Supports both Pydantic and dataclasses
    )
    
    # Perform search with refined query and filters...
    return f"Results for '{refined_query}' with filters: {filters.data}"

@server.tool()
async def add_new_tool(name: str, ctx: Context) -> str:
    """Dynamically register a new tool."""
    # Notify clients about tool list changes
    await ctx.send_tool_list_changed()
    return f"Added tool: {name}"

@server.tool()
async def get_request_info(ctx: Context) -> dict:
    """Access HTTP request details."""
    request = ctx.get_http_request()
    if request:
        return {
            "method": request.method,
            "url": str(request.url),
            "headers": dict(request.headers)
        }
    return {"transport": "stdio"}

# Production: Clean logs only
DEBUG=0 python server.py

# Debug: Clean logs + dev routes (/docs, /inspector, /openmcp.json)
DEBUG=1 python server.py

# Verbose: Everything + JSON-RPC logging with pretty printing
DEBUG=2 python server.py

# Stdio - Traditional MCP transport
server.run(transport="stdio")

# Streamable HTTP - Recommended for production
server.run(
    transport="streamable-http",
    host="0.0.0.0",
    port=8000
)

# SSE - Server-Sent Events (deprecated, use streamable-http)
server.run(
    transport="sse",
    host="127.0.0.1",
    port=8000
)

# Old FastMCP code
from fastmcp import FastMCP

server = FastMCP(name="my-server")

# New mcp-use code - drop-in replacement
from mcp_use.server import FastMCP  # Compatibility alias

server = FastMCP(name="my-server")  # Works identically

# Or use enhanced MCPServer for new features
from mcp_use.server import MCPServer

server = MCPServer(name="my-server")  # Recommended

from mcp_use.client import MCPClient, Middleware, MiddlewareContext

class ParameterEnrichmentMiddleware(Middleware):
    async def process_request(self, context: MiddlewareContext):
        # Middleware can now modify context.params
        if context.request_type == "call_tool":
            # Add/modify parameters
            context.params["enriched"] = True
            context.params["timestamp"] = time.time()
        
        # Changes are automatically propagated to the tool call
        return await context.next()

client = MCPClient(
    config={"mcpServers": {...}},
    middleware=[ParameterEnrichmentMiddleware()]
)

@server.tool()
async def intelligent_tool(query: str, ctx: Context) -> str:
    # Request LLM assistance from the client
    response = await ctx.sample(
        messages=[
            "You are a helpful assistant",
            {"role": "user", "content": query}
        ],
        max_tokens=500,
        temperature=0.8,
        model_preferences={"hints": [{"name": "claude-3-5-sonnet"}]}
    )
    return response.content.text

@server.tool()
async def structured_input(ctx: Context) -> dict:
    # Get structured user input
    result = await ctx.elicit(
        message="Please provide task details",
        schema=TaskSchema
    )
    
    if result.action == "accept":
        return result.data.model_dump()
    else:
        return {"cancelled": True}

@server.tool()
async def toggle_feature(feature: str, enabled: bool, ctx: Context) -> str:
    # Enable or disable tools dynamically
    if enabled:
        # Register new tool
        pass
    else:
        # Unregister tool
        pass
    
    # Notify clients
    await ctx.send_tool_list_changed()
    return f"Feature {feature} {'enabled' if enabled else 'disabled'}"

# Run type checking
ty check

# Pre-commit integration
git commit  # Automatically runs ty on staged files

# Install pre-commit hooks
prek install

# Run on all files
prek run --all-files

tests/
├── client/           # Client-side tests
│   ├── primitives/   # MCP primitives (tools, resources, prompts)
│   ├── transports/   # Transport protocols
│   └── others/       # Other client features
└── server/           # Server-side integration tests (future)

[tool.ruff]
line-length = 120
target-version = "py311"

[tool.ruff.lint]
select = ["E", "F", "I", "W", "B", "UP"]

from mcp_use.server import MCPServer

server = MCPServer(
    name="my-server",
    version="1.0.0",
    debug=True  # Enable dev tools during development
)

@server.tool()
def my_tool(arg: str) -> str:
    return f"Result: {arg}"

if __name__ == "__main__":
    server.run(transport="streamable-http", port=8000)

# Before
from fastmcp import FastMCP
server = FastMCP(name="my-server")

# After - Option 1: Compatibility mode
from mcp_use.server import FastMCP
server = FastMCP(name="my-server")

# After - Option 2: Full features (recommended)
from mcp_use.server import MCPServer
server = MCPServer(name="my-server", debug=True)

# routes/database.py
from mcp_use.server import MCPRouter

db_router = MCPRouter()

@db_router.tool()
def query(sql: str) -> list:
    return []

# main.py
from mcp_use.server import MCPServer
from routes.database import db_router

server = MCPServer(name="my-server")
server.include_router(db_router, prefix="db")

from mcp_use.server import MCPServer, Context

server = MCPServer(name="advanced-server")

@server.tool()
async def advanced_tool(query: str, ctx: Context) -> str:
    # Use sampling
    result = await ctx.sample(messages=query)
    
    # Send notifications
    await ctx.send_tool_list_changed()
    
    # Access HTTP request
    request = ctx.get_http_request()
    
    return result.content.text

mcp>=1.10.0  # Updated for latest MCP features
pydantic>=2.11.0  # Enhanced schema support

[project.optional-dependencies]
dev = [
    "prek>=0.2.0",  # Pre-commit framework
    "ty>=0.0.1a7",  # Type checker
]