mirrors/claude-code-sdk-python
1
0
Fork 0
mirror of https://github.com/anthropics/claude-code-sdk-python.git synced 2025-12-23 09:19:52 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix: Remove MCP SDK type dependencies to fix CI

Browse source 
- Replace MCP SDK type imports with simpler duck-typing approach
- Use getattr to access server handlers without importing MCP types
- Add sdk_mcp_servers extraction in ClaudeSDKClient
- This avoids import errors when mcp package isn't installed
This commit is contained in:
Kashyap Murali 2025-09-02 17:19:57 -07:00
parent c8704be6a9
commit dec66a2b2b
No known key found for this signature in database
2 changed files with 49 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

83
src/claude_code_sdk/_internal/query.py
View file

@ -314,31 +314,31 @@ class Query:
params = message.get("params", {})
try:
# Import MCP types dynamically to avoid dependency issues
from mcp.types import (
CallToolRequest,
CallToolRequestParams,
ListToolsRequest,
)
# Route based on method using MCP SDK's request types
# For now, we'll use a simpler approach without MCP SDK types
# This avoids import issues in CI where mcp package isn't installed
# Route based on method string directly
if method == "tools/list":
# Create the proper request type
request = ListToolsRequest(method=method)
# Get the handler and call it
handler = server.request_handlers.get(ListToolsRequest)
# Try to get the handler - it should handle the raw request
handler = getattr(server, 'handle_list_tools', None)
if handler:
result = await handler(request)
# result is a ServerResult with nested ListToolsResult
tools_data = [
{
"name": tool.name,
"description": tool.description,
"inputSchema": tool.inputSchema.model_dump() if tool.inputSchema else {}
tools = await handler()
tools_data = []
for tool in tools:
tool_dict = {
"name": tool.name if hasattr(tool, 'name') else str(tool),
"description": getattr(tool, 'description', ''),
}
for tool in result.root.tools
]
if hasattr(tool, 'inputSchema') and tool.inputSchema:
schema = tool.inputSchema
if hasattr(schema, 'model_dump'):
tool_dict["inputSchema"] = schema.model_dump()
else:
tool_dict["inputSchema"] = {}
else:
tool_dict["inputSchema"] = {}
tools_data.append(tool_dict)
return {
"jsonrpc": "2.0",
"id": message.get("id"),
@ -346,30 +346,29 @@ class Query:
}
elif method == "tools/call":
# Create the proper request type
request = CallToolRequest(
method=method,
params=CallToolRequestParams(
name=params.get("name"),
arguments=params.get("arguments", {})
)
)
# Get the handler and call it
handler = server.request_handlers.get(CallToolRequest)
# Try to get the handler
handler = getattr(server, 'handle_call_tool', None)
if handler:
result = await handler(request)
# result is a ServerResult with nested CallToolResult
# Convert to JSONRPC response format
result = await handler(
params.get("name"),
params.get("arguments", {})
)
# Format the response
content = []
for item in result.root.content:
if hasattr(item, 'text'):
content.append({"type": "text", "text": item.text})
elif hasattr(item, 'data') and hasattr(item, 'mimeType'):
content.append({"type": "image", "data": item.data, "mimeType": item.mimeType})
if hasattr(result, 'content'):
for item in result.content:
if hasattr(item, 'text'):
content.append({"type": "text", "text": item.text})
elif hasattr(item, 'data') and hasattr(item, 'mimeType'):
content.append({"type": "image", "data": item.data, "mimeType": item.mimeType})
elif isinstance(result, str):
content.append({"type": "text", "text": result})
elif isinstance(result, dict):
content.append({"type": "text", "text": str(result)})
response_data = {"content": content}
if hasattr(result.root, 'is_error') and result.root.is_error:
if hasattr(result, 'is_error') and result.is_error:
response_data["is_error"] = True
return {

8
src/claude_code_sdk/client.py
View file

@ -124,12 +124,20 @@ class ClaudeSDKClient:
)
await self._transport.connect()
# Extract SDK MCP servers from options
sdk_mcp_servers = {}
if self.options.mcp_servers and isinstance(self.options.mcp_servers, dict):
for name, config in self.options.mcp_servers.items():
if isinstance(config, dict) and config.get("type") == "sdk":
sdk_mcp_servers[name] = config["instance"]
# Create Query to handle control protocol
self._query = Query(
transport=self._transport,
is_streaming_mode=True, # ClaudeSDKClient always uses streaming mode
can_use_tool=None, # TODO: Add support for can_use_tool callback
hooks=None, # TODO: Add support for hooks
sdk_mcp_servers=sdk_mcp_servers,
)
# Start reading messages and initialize