import asyncio
import mcp.types as types
from mcp_use import MCPClient
# A dedicated handler for log messages
async def handle_logs(log_params: types.LoggingMessageNotificationParams):
print(f"LOG [{log_params.level.upper()}]: {log_params.message}")
async def test_logging(primitive_server):
"""Tests receiving logs from the primitive server."""
config = {"mcpServers": {"PrimitiveServer": {"url": f"{primitive_server}/mcp"}}}
# Pass the callback to the client
client = MCPClient(config, logging_callback=handle_logs)
try:
await client.create_all_sessions()
session = client.get_session("PrimitiveServer")
# This tool will trigger the logging_callback
result = await session.call_tool(name="logging_tool", arguments={})
assert result.content[0].text == "Logging tool completed"
finally:
await client.close_all_sessions()