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
| 要点 | EN | CN |
|---|---|---|
@mcp.resource() 注册资源提供者 | @mcp.resource() registers a resource provider function | @mcp.resource() 注册资源提供者函数 |
| Resource = 只读数据（GET 模式） | Resource = read-only data (GET model) | Resource = 只读数据（GET 模式） |
| Tool = 有副作用的操作（POST 模式） | Tool = action with side effects (POST model) | Tool = 有副作用的操作（POST 模式） |
URI 模板用 {参数} 实现动态资源 | URI templates use {param} for dynamic resources | URI 模板用 {param} 实现动态资源 |
| 资源未找到用错误码 -32002 | Resource not found uses error code -32002 | 资源未找到用错误码 -32002 |
所有 MCP 错误用 McpError 类 | Use McpError class for all MCP errors | 所有 MCP 错误用 McpError 类 |
| 不要向客户端暴露内部错误 | Don’t expose internal errors to clients | 不要向客户端暴露内部错误 |
| 错误后需清理资源 | Clean up resources after errors | 错误后需清理资源 |
| 用 try-catch 包装工具/资源调用 | Wrap tool/resource calls with try-catch | 用 try-catch 包装工具/资源调用 |
| 记录安全相关错误 | Log security-relevant errors | 记录安全相关错误 |