लॉगिंग

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"})

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]}")

        # एक tool कॉल करें - लॉग्स हैंडलर के माध्यम से दिखाई देंगे
        result = await client.call_tool("analyze", {"input": "test"})
        print(f"Result: {result.content}")

asyncio.run(main())