aidputils.agents.tools.mcp.mcp_service¶
- aidputils.agents.tools.mcp.mcp_service.get_mcp_client(server_name: str, server_url: str, auth: dict | None = None, transport: str = 'streamable_http', custom_headers: Dict[str, str] | None = None)[source]¶
Build and return an MCPHTTPClient for the given server_url with optional auth handling.
- Auth handling:
NO_AUTH (default): no Authorization header is set
BEARER_TOKEN: requires ‘token’ in auth dict; adds Authorization: Bearer <token>
Unsupported auth types will raise ValueError.
- async aidputils.agents.tools.mcp.mcp_service.list_objects(client, runtime_params: Dict[str, Any])[source]¶
Generic dispatcher to list or fetch MCP objects based on runtime_params.operation.
- Supported operations:
list_tools: paginated listing of tools
list_resources: paginated listing of resources
get_prompt: fetch a prompt by name/id with optional variables
- async aidputils.agents.tools.mcp.mcp_service.list_objects_paginated(client, runtime_params: Dict[str, Any])[source]¶
Wrapper around list_objects that normalizes pagination responses for list_tools and list_resources. Non-paginated ops (e.g., get_prompt) pass through the underlying result.
- async aidputils.agents.tools.mcp.mcp_service.list_all_objects(client, runtime_params: Dict[str, Any])[source]¶
- aidputils.agents.tools.mcp.mcp_service.list_objects_paginated_sync(client, runtime_params: Dict[str, Any])[source]¶
Synchronous wrapper for list_objects_paginated for use in sync code paths.