5.3 配置选项

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    name="My MCP Server",          # 服务器名称
    description="服务器描述",        # 服务器描述
    version="1.0.0",               # 服务器版本
    verbose=True,                  # 是否启用详细日志
    base_path="/api/mcp"           # API基本路径（用于HTTP传输）
)

if __name__ == "__main__":
    mcp.run(
        transport="stdio",
        buffer_size=4096        # 缓冲区大小（字节）
    )

if __name__ == "__main__":
    mcp.run(
        transport="sse",
        host="0.0.0.0",         # 监听地址
        port=8000,              # 监听端口
        cors_origins=["*"],     # CORS允许的源
        ssl_cert="cert.pem",    # SSL证书路径（可选）
        ssl_key="key.pem"       # SSL密钥路径（可选）
    )

if __name__ == "__main__":
    mcp.run(
        transport="websocket",
        host="0.0.0.0",         # 监听地址
        port=8000,              # 监听端口
        cors_origins=["*"],     # CORS允许的源
        ssl_cert="cert.pem",    # SSL证书路径（可选）
        ssl_key="key.pem",      # SSL密钥路径（可选）
        ping_interval=20        # 心跳间隔（秒）
    )

if __name__ == "__main__":
    mcp.run(
        transport="http",
        host="0.0.0.0",         # 监听地址
        port=8000,              # 监听端口
        cors_origins=["*"],     # CORS允许的源
        ssl_cert="cert.pem",    # SSL证书路径（可选）
        ssl_key="key.pem"       # SSL密钥路径（可选）
    )

@mcp.resource("data://cached/{id}", cache_ttl=300)
def get_cached_data(id: str) -> dict:
    """获取缓存的数据"""
    # 实现...

@mcp.tool(timeout=30, rate_limit=10)
async def process_data(data: dict) -> dict:
    """处理数据"""
    # 实现...

from mcp.server.fastmcp import FastMCP, Auth

# 定义认证回调
def authenticate(token: str) -> bool:
    return token == "secret-token"

mcp = FastMCP(
    name="Secure MCP Server",
    auth=Auth(
        enabled=True,
        authenticate=authenticate,
        token_header="X-API-Token"
    )
)

mcp = FastMCP(
    name="High Performance Server",
    worker_threads=16,           # 工作线程数
    max_connection_queue=100,    # 最大连接队列
    resource_cache_size=1024,    # 资源缓存大小
    response_compression=True    # 启用响应压缩
)

from mcp import ClientSession, ClientConfig
from mcp.client.sse import sse_client

async def run():
    config = ClientConfig(
        connect_timeout=5.0,     # 连接超时（秒）
        request_timeout=30.0,    # 请求超时（秒）
        max_redirects=5          # 最大重定向次数
    )
    
    async with sse_client(
        "http://localhost:8000/mcp",
        headers={"User-Agent": "MCP Client/1.0"}
    ) as (read, write):
        async with ClientSession(read, write, config=config) as session:
            # 使用会话...

config = ClientConfig(
    auto_reconnect=True,     # 自动重连
    reconnect_attempts=3,    # 重连尝试次数
    reconnect_delay=1.0,     # 重连延迟（秒）
    session_ttl=3600         # 会话生存时间（秒）
)

config = ClientConfig(
    retry_on_error=True,        # 错误时重试
    max_retries=3,              # 最大重试次数
    retry_delay=0.5,            # 重试延迟（秒）
    exponential_backoff=True    # 指数退避
)

config = ClientConfig(
    request_timeout=30.0,        # 请求超时（秒）
    tool_timeout=60.0,           # 工具调用超时（秒）
    resource_timeout=15.0,       # 资源读取超时（秒）
    prompt_timeout=5.0           # 提示渲染超时（秒）
)

# 为服务器设置环境变量
export MCP_SERVER_PORT=9000
export MCP_SERVER_TRANSPORT=sse
export MCP_SERVER_VERBOSE=true

# 启动服务器
python my_server.py

# 为客户端设置环境变量
export MCP_CLIENT_SERVER_URL=http://localhost:9000/mcp
export MCP_CLIENT_REQUEST_TIMEOUT=60.0
export MCP_AUTH_TOKEN=my-secret-token

# 运行客户端
python my_client.py

import json
import os

def load_config(config_path="config.json"):
    # 首先使用默认配置
    config = {
        "server": {
            "port": 8000,
            "host": "127.0.0.1",
            "transport": "sse",
            "worker_threads": 4
        },
        "security": {
            "auth_enabled": False,
            "ssl_enabled": False
        }
    }
    
    # 从文件加载配置
    if os.path.exists(config_path):
        with open(config_path, "r") as f:
            file_config = json.load(f)
            # 递归更新配置
            def update_config(target, source):
                for key, value in source.items():
                    if key in target and isinstance(target[key], dict) and isinstance(value, dict):
                        update_config(target[key], value)
                    else:
                        target[key] = value
                        
            update_config(config, file_config)
    
    # 环境变量覆盖配置
    if os.environ.get("MCP_SERVER_PORT"):
        config["server"]["port"] = int(os.environ["MCP_SERVER_PORT"])
    
    return config

# 使用配置
config = load_config()
mcp = FastMCP(
    name="Configured Server",
    verbose=config["server"].get("verbose", False)
)

if __name__ == "__main__":
    mcp.run(
        transport=config["server"]["transport"],
        host=config["server"]["host"],
        port=config["server"]["port"],
        worker_threads=config["server"]["worker_threads"]
    )

import os

# 获取当前环境
env = os.environ.get("MCP_ENV", "development")

# 加载配置
config = load_config(f"config/{env}.json")

# 生产环境下的配置
if env == "production":
    # 启用SSL
    mcp.run(
        transport="sse",
        host="0.0.0.0",
        port=443,
        ssl_cert="/path/to/cert.pem",
        ssl_key="/path/to/key.pem"
    )
    
    # 启用身份验证
    mcp = FastMCP(
        name="Production Server",
        auth=Auth(
            enabled=True,
            authenticate=lambda token: validate_token_securely(token),
            token_header="X-API-Token"
        )
    )

# 高性能服务器配置
mcp = FastMCP(
    name="High Performance Server",
    worker_threads=os.cpu_count() * 2,  # CPU核心数的两倍
    resource_cache_size=10000,          # 大资源缓存
    response_compression=True           # 启用压缩
)

# 配置资源缓存
@mcp.resource("data://frequently-accessed/{id}", cache_ttl=3600)
def get_frequent_data(id: str) -> dict:
    """获取频繁访问的数据，缓存1小时"""
    # 实现...

import logging
from logging.handlers import RotatingFileHandler

# 配置日志
log_handler = RotatingFileHandler(
    'mcp_server.log',           # 日志文件
    maxBytes=10*1024*1024,      # 10MB
    backupCount=5               # 保留5个备份
)

logging.basicConfig(
    level=logging.INFO if env == "production" else logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[log_handler]
)

# 在服务器中使用
mcp = FastMCP(
    name="Monitored Server",
    verbose=env != "production"  # 仅在非生产环境启用详细日志
)

@mcp.on_initialize
async def on_server_start(ctx):
    ctx.logger.info(f"服务器已启动，环境: {env}")

@mcp.on_shutdown
async def on_server_stop(ctx):
    ctx.logger.info("服务器已关闭")