跳转到主要内容
服务器可以在 tool 执行期间向客户端发送日志信息，从而提高对服务器运行状况的可见性，并帮助进行调试。

​
处理程序

from dedalus_mcp.client import ClientCapabilitiesConfig, open_connection
from dedalus_mcp.types import LoggingMessageNotificationParams

def logging_handler(params: LoggingMessageNotificationParams) -> None:
    """处理来自服务器的日志消息。"""
    level = params.level.upper()
    data = params.data

    print(f"[{level}] {data}")

​
使用方式

capabilities = ClientCapabilitiesConfig(logging=logging_handler)

async with open_connection(
    url="http://127.0.0.1:8000/mcp",
    transport="streamable-http",
    capabilities=capabilities,
) as client:
    # 服务器日志将路由到您的处理程序
    result = await client.call_tool("process", {"file": "data.csv"})

​
日志级别

服务器可以按不同级别发送日志：
级别说明
debug详细的调试信息
info一般运行信息
warning警告条件
error错误条件

​
结构化日志

将其接入你的日志框架： 
import logging

logger = logging.getLogger("mcp.server")

def logging_handler(params: LoggingMessageNotificationParams) -> None:
    level = getattr(logging, params.level.upper(), logging.INFO)
    logger.log(level, params.data)

​
完整示例

import asyncio
from dedalus_mcp.client import ClientCapabilitiesConfig, open_connection
from dedalus_mcp.types import LoggingMessageNotificationParams

def logging_handler(params: LoggingMessageNotificationParams) -> None:
    level = params.level.upper()
    print(f"[SERVER {level}] {params.data}")

async def main():
    capabilities = ClientCapabilitiesConfig(logging=logging_handler)

    async with open_connection(
        url="http://127.0.0.1:8000/mcp",
        transport="streamable-http",
        capabilities=capabilities,
    ) as client:
        print(f"Connected: {client.initialize_result.serverInfo.name}")

        tools = await client.list_tools()
        print(f"Available tools: {[t.name for t in tools.tools]}")

        # 调用工具 - 日志将通过处理器显示
        result = await client.call_tool("analyze", {"input": "test"})
        print(f"Result: {result.content}")

asyncio.run(main())