Category: AI/LLM

The @app.resource() decorator in MCP Python SDK marks a function as a resource provider—a function that returns data (text, JSON, binary) when the AI requests a specific URI.

CN: MCP Python SDK 中的 @app.resource() 装饰器，把一个函数标记为资源提供者——当 AI 请求某个特定 URI 时，这个函数返回数据（文本、JSON、二进制）。

The @app.resource() decorator registers a function as a resource handler. The function must return either a string, a types.TextContent, or a types.Resource object.

CN: @app.resource() 装饰器把一个函数注册为资源处理器。函数必须返回字符串、types.TextContent 或 types.Resource 对象。

Code Implementation

# ============================================================
# MCP Server with @app.resource() and Error Handling
# 带 @app.resource() 和错误处理的 MCP 服务器
# ============================================================

import asyncio
import json
import logging
from typing import Optional, List, Dict, Any

# MCP SDK imports / MCP SDK 导入
from mcp.server.fastmcp import FastMCP
import mcp.types as types
from mcp.shared.exceptions import McpError

# Configure logging / 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ============================================================
# 1. Create MCP Server instance / 创建 MCP 服务器实例
# ============================================================
# FastMCP is the high-level server class that handles all the protocol details
# FastMCP 是高层服务器类，处理所有协议细节
mcp = FastMCP(
    name="data-engineering-mcp-server",
    version="1.0.0",
    description="MCP Server for Data Engineering tasks"
)

# ============================================================
# 2. STATIC RESOURCE: Fixed URI, always returns same data
#    静态资源：固定 URI，始终返回相同数据
# ============================================================
# @mcp.resource() registers this function as a resource provider
# @mcp.resource() 把这个函数注册为资源提供者
# The URI "config://settings" is the address the AI uses to access this resource
# URI "config://settings" 是 AI 用来访问此资源的地址
@mcp.resource("config://settings")
async def get_config() -> str:
    """
    Returns static configuration settings for the AI.
    返回 AI 的静态配置设置。
    
    This is a simple static resource - it always returns the same JSON.
    这是一个简单的静态资源——它总是返回相同的 JSON。
    """
    # Return as a JSON string / 以 JSON 字符串返回
    return json.dumps({
        "max_rows_return": 1000,
        "default_database": "analytics_prod",
        "query_timeout_seconds": 30,
        "enable_cache": True
    }, indent=2)


# ============================================================
# 3. STATIC RESOURCE with TextContent (more control)
#    带 TextContent 的静态资源（更多控制）
# ============================================================
@mcp.resource("info://version")
async def get_version_info() -> types.TextContent:
    """
    Returns version information as TextContent with MIME type.
    以 TextContent 返回版本信息，带 MIME 类型。
    
    TextContent allows you to specify the MIME type of the response.
    TextContent 允许你指定响应的 MIME 类型。
    """
    return types.TextContent(
        type="text",
        text="MCP Data Engineering Server v1.0.0 (built 2026-07-25)",
        # MIME type tells the client how to interpret the content
        # MIME 类型告诉客户端如何解释内容
        mime_type="text/plain"
    )


# ============================================================
# 4. DYNAMIC RESOURCE (Resource Template) with parameter
#    动态资源（资源模板）带参数
# ============================================================
# URI template with {table_name} placeholder - the AI provides the actual value
# URI 模板带 {table_name} 占位符——AI 提供实际值
# Example: AI requests "schema://tables/orders" -> table_name = "orders"
# 例如：AI 请求 "schema://tables/orders" -> table_name = "orders"
@mcp.resource("schema://tables/{table_name}")
async def get_table_schema(table_name: str) -> str:
    """
    Returns the schema of a specific database table.
    返回特定数据库表的 schema。
    
    This is a dynamic resource - the result depends on the {table_name} parameter.
    这是一个动态资源——结果取决于 {table_name} 参数。
    
    Args:
        table_name: Name of the table to describe / 要描述的表名
    """
    # In production, this would query your database metadata
    # 在生产环境中，这会查询你的数据库元数据
    # For demo, we return mock schema data / 演示用，返回模拟 schema 数据
    
    # Simulate a database schema lookup / 模拟数据库 schema 查询
    mock_schemas: Dict[str, Dict[str, Any]] = {
        "orders": {
            "columns": [
                {"name": "order_id", "type": "BIGINT", "nullable": False},
                {"name": "customer_id", "type": "BIGINT", "nullable": False},
                {"name": "order_date", "type": "DATE", "nullable": False},
                {"name": "total_amount", "type": "DECIMAL(10,2)", "nullable": False},
                {"name": "status", "type": "VARCHAR(50)", "nullable": False},
            ],
            "primary_key": "order_id",
            "partitioned_by": "order_date"
        },
        "customers": {
            "columns": [
                {"name": "customer_id", "type": "BIGINT", "nullable": False},
                {"name": "customer_name", "type": "VARCHAR(200)", "nullable": False},
                {"name": "email", "type": "VARCHAR(255)", "nullable": True},
                {"name": "created_at", "type": "TIMESTAMP", "nullable": False},
            ],
            "primary_key": "customer_id"
        }
    }
    
    # ============================================================
    # ERROR HANDLING: Resource not found
    # 错误处理：资源未找到
    # ============================================================
    # If the requested table doesn't exist, raise an McpError
    # 如果请求的表不存在，抛出 McpError
    if table_name not in mock_schemas:
        # McpError is the standard MCP error type
        # McpError 是标准的 MCP 错误类型
        # -32002 is the standard error code for "Resource Not Found"[reference:28]
        # -32002 是 "资源未找到" 的标准错误码[reference:29]
        raise McpError(
            code=-32002,  # Resource Not Found / 资源未找到
            message=f"Table '{table_name}' not found in database schema. "
                    f"Available tables: {', '.join(mock_schemas.keys())}"
        )
    
    # Return the schema as formatted JSON / 以格式化的 JSON 返回 schema
    return json.dumps(mock_schemas[table_name], indent=2)


# ============================================================
# 5. DYNAMIC RESOURCE with multiple parameters
#    带多个参数的动态资源
# ============================================================
@mcp.resource("query://results/{query_id}/{format}")
async def get_query_results(query_id: str, format: str) -> str:
    """
    Returns cached query results in specified format.
    以指定格式返回缓存的查询结果。
    
    This demonstrates a resource with TWO parameters.
    这演示了带两个参数的资源。
    
    Args:
        query_id: The ID of the saved query / 保存的查询的 ID
        format: Output format: 'json' or 'csv' / 输出格式：'json' 或 'csv'
    """
    # In production, this would fetch from a cache or database
    # 在生产环境中，这会从缓存或数据库获取
    
    # Simulate query results / 模拟查询结果
    mock_results = [
        {"date": "2026-07-01", "revenue": 125000, "orders": 342},
        {"date": "2026-07-02", "revenue": 98000, "orders": 287},
        {"date": "2026-07-03", "revenue": 156000, "orders": 415},
    ]
    
    # ============================================================
    # ERROR HANDLING: Invalid parameter value
    # 错误处理：无效参数值
    # ============================================================
    if format not in ["json", "csv"]:
        # -32602 is "Invalid params" / -32602 是 "无效参数"
        raise McpError(
            code=-32602,
            message=f"Invalid format '{format}'. Supported formats: json, csv"
        )
    
    if format == "csv":
        # Convert to CSV format / 转换为 CSV 格式
        import csv
        from io import StringIO
        output = StringIO()
        writer = csv.DictWriter(output, fieldnames=mock_results[0].keys())
        writer.writeheader()
        writer.writerows(mock_results)
        return output.getvalue()
    
    # Default: JSON format / 默认：JSON 格式
    return json.dumps(mock_results, indent=2)


# ============================================================
# 6. RESOURCE with list of resources (for client discovery)
#    带资源列表的资源（用于客户端发现）
# ============================================================
# @mcp.list_resources() lets the client discover all available resources
# @mcp.list_resources() 让客户端发现所有可用的资源
# This is called automatically by the SDK, but you can customize it
# 这是由 SDK 自动调用的，但你可以自定义它
@mcp.list_resources()
async def list_resources() -> List[types.Resource]:
    """
    Returns the list of all resources this server provides.
    返回此服务器提供的所有资源的列表。
    
    The client calls this to discover what resources are available.
    客户端调用此方法来发现有哪些资源可用。
    """
    return [
        types.Resource(
            uri="config://settings",
            name="Configuration Settings",
            description="Server configuration settings",
            mimeType="application/json"
        ),
        types.Resource(
            uri="info://version",
            name="Version Information",
            description="Server version info",
            mimeType="text/plain"
        ),
        # Resource template - uses URI template notation
        # 资源模板 - 使用 URI 模板表示法
        types.Resource(
            uri="schema://tables/{table_name}",
            name="Table Schema",
            description="Schema of a specific database table",
            mimeType="application/json"
        ),
        types.Resource(
            uri="query://results/{query_id}/{format}",
            name="Query Results",
            description="Cached query results in JSON or CSV format",
            mimeType="application/json"
        ),
    ]


# ============================================================
# 7. TOOL with error handling (for comparison)
#    带错误处理的 TOOL（用于对比）
# ============================================================
@mcp.tool()
async def execute_sql_query(query: str, limit: Optional[int] = None) -> str:
    """
    Execute a SQL query and return results.
    执行 SQL 查询并返回结果。
    
    This is a TOOL (action) for comparison with RESOURCES (data).
    这是一个 TOOL（动作），用于与 RESOURCES（数据）对比。
    """
    # ============================================================
    # ERROR HANDLING: Tool-specific errors
    # 错误处理：Tool 特定的错误
    # ============================================================
    # Tools should also use McpError for consistent error handling[reference:30]
    # Tools 也应该使用 McpError 来实现一致的错误处理[reference:31]
    
    # Validate input / 验证输入
    if not query or len(query.strip()) == 0:
        raise McpError(
            code=-32602,  # Invalid params / 无效参数
            message="SQL query cannot be empty"
        )
    
    # Check for dangerous SQL (simple demo check)
    # 检查危险的 SQL（简单演示检查）
    dangerous_keywords = ["DROP", "TRUNCATE", "DELETE", "UPDATE", "ALTER"]
    upper_query = query.upper()
    for keyword in dangerous_keywords:
        if keyword in upper_query:
            raise McpError(
                code=-32603,  # Internal error / 内部错误
                message=f"Query contains dangerous keyword '{keyword}'. Only SELECT queries are allowed."
            )
    
    # Simulate query execution / 模拟查询执行
    # In production, this would connect to your database
    # 在生产环境中，这会连接你的数据库
    mock_result = {
        "query": query,
        "row_count": 42,
        "execution_time_ms": 125,
        "results": [
            {"id": 1, "value": "sample data 1"},
            {"id": 2, "value": "sample data 2"},
        ][:limit if limit else 2]
    }
    
    return json.dumps(mock_result, indent=2)


# ============================================================
# 8. GLOBAL ERROR HANDLER (optional but recommended)
#    全局错误处理器（可选但推荐）
# ============================================================
# You can wrap resource functions with a decorator for consistent error handling
# 你可以用装饰器包装资源函数，实现一致的错误处理
def handle_resource_errors(func):
    """
    Decorator that catches exceptions and converts them to McpError.
    装饰器，捕获异常并转换为 McpError。
    
    This ensures all resource errors are returned in the MCP standard format.
    这确保所有资源错误都以 MCP 标准格式返回。
    """
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except McpError:
            # McpError is already in the correct format - re-raise
            # McpError 已经是正确格式 - 重新抛出
            raise
        except Exception as e:
            # Unexpected errors - convert to McpError with internal error code
            # 意外错误 - 转换为带内部错误码的 McpError
            logger.error(f"Unexpected error in resource {func.__name__}: {e}")
            raise McpError(
                code=-32603,  # Internal error / 内部错误
                message=f"An unexpected error occurred: {str(e)}"
            )
    return wrapper


# ============================================================
# 9. SERVER ENTRY POINT / 服务器入口点
# ============================================================
if __name__ == "__main__":
    # Run the server with stdio transport (for local communication)
    # 使用 stdio 传输运行服务器（用于本地通信）
    # This is the most common way to run MCP servers[reference:32]
    # 这是运行 MCP 服务器最常见的方式[reference:33]
    mcp.run(transport="stdio")

Key Takeaways