wehub-resource-sync
49b9bb6724
Deploy Docs / deploy-docs (push) Failing after 1s
Conformance Tests / client-conformance (push) Failing after 3s
Conformance Tests / server-conformance (push) Failing after 1s
GitHub Actions Security Analysis / zizmor (push) Failing after 1s
CI / checks (push) Failing after 59m20s
CI / all-green (push) Waiting to run
63 lines
2.5 KiB
Python
63 lines
2.5 KiB
Python
"""examples/snippets/clients/parsing_tool_results.py"""
|
|
|
|
import asyncio
|
|
|
|
import mcp_types as types
|
|
|
|
from mcp import ClientSession, StdioServerParameters
|
|
from mcp.client.stdio import stdio_client
|
|
|
|
|
|
async def parse_tool_results():
|
|
"""Demonstrates how to parse different types of content in CallToolResult."""
|
|
server_params = StdioServerParameters(command="python", args=["path/to/mcp_server.py"])
|
|
|
|
async with stdio_client(server_params) as (read, write):
|
|
async with ClientSession(read, write) as session:
|
|
await session.initialize()
|
|
|
|
# Example 1: Parsing text content
|
|
result = await session.call_tool("get_data", {"format": "text"})
|
|
for content in result.content:
|
|
if isinstance(content, types.TextContent):
|
|
print(f"Text: {content.text}")
|
|
|
|
# Example 2: Parsing structured content from JSON tools
|
|
result = await session.call_tool("get_user", {"id": "123"})
|
|
if hasattr(result, "structured_content") and result.structured_content:
|
|
# Access structured data directly
|
|
user_data = result.structured_content
|
|
print(f"User: {user_data.get('name')}, Age: {user_data.get('age')}")
|
|
|
|
# Example 3: Parsing embedded resources
|
|
result = await session.call_tool("read_config", {})
|
|
for content in result.content:
|
|
if isinstance(content, types.EmbeddedResource):
|
|
resource = content.resource
|
|
if isinstance(resource, types.TextResourceContents):
|
|
print(f"Config from {resource.uri}: {resource.text}")
|
|
else:
|
|
print(f"Binary data from {resource.uri}")
|
|
|
|
# Example 4: Parsing image content
|
|
result = await session.call_tool("generate_chart", {"data": [1, 2, 3]})
|
|
for content in result.content:
|
|
if isinstance(content, types.ImageContent):
|
|
print(f"Image ({content.mime_type}): {len(content.data)} bytes")
|
|
|
|
# Example 5: Handling errors
|
|
result = await session.call_tool("failing_tool", {})
|
|
if result.is_error:
|
|
print("Tool execution failed!")
|
|
for content in result.content:
|
|
if isinstance(content, types.TextContent):
|
|
print(f"Error: {content.text}")
|
|
|
|
|
|
async def main():
|
|
await parse_tool_results()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|