MCP Client API Reference¶
Client libraries for connecting to MCP servers.
MCPClient¶
promptise.mcp.client.MCPClient
¶
Production-grade MCP client for a single server.
Supports HTTP (Streamable HTTP), SSE, and stdio transports with Bearer token or API key authentication.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str | None
|
Server endpoint URL (for HTTP/SSE). Mutually exclusive
with |
None
|
transport
|
str
|
|
'http'
|
headers
|
dict[str, str] | None
|
Extra HTTP headers sent on every request. |
None
|
bearer_token
|
str | None
|
Pre-issued Bearer token. When provided, an
|
None
|
api_key
|
str | None
|
Pre-shared API key. When provided, an |
None
|
command
|
str | None
|
Executable for stdio transport (e.g. |
None
|
args
|
list[str] | None
|
Arguments for the stdio command. |
None
|
env
|
dict[str, str] | None
|
Environment variables for the stdio process. |
None
|
timeout
|
float
|
HTTP request timeout in seconds. |
30.0
|
Example — unauthenticated::
async with MCPClient(url="http://localhost:8080/mcp") as client:
tools = await client.list_tools()
Example — with Bearer token::
async with MCPClient(
url="http://localhost:8080/mcp",
bearer_token="eyJhbGciOiJIUzI1NiIs...",
) as client:
result = await client.call_tool("search", {"query": "python"})
Example — with API key::
async with MCPClient(
url="http://localhost:8080/mcp",
api_key="my-secret-key",
) as client:
tools = await client.list_tools()
Example — fetch token from server endpoint::
token = await MCPClient.fetch_token(
"http://localhost:8080/auth/token",
client_id="my-agent",
client_secret="agent-secret",
)
async with MCPClient(
url="http://localhost:8080/mcp",
bearer_token=token,
) as client:
tools = await client.list_tools()
headers
property
¶
Current HTTP headers (read-only copy).
session
property
¶
The underlying MCP ClientSession, or None if not connected.
__aenter__()
async
¶
Connect to the server and initialise the session.
__aexit__(*exc)
async
¶
Close the session and transport.
Suppresses CancelledError during cleanup — the MCP SDK's
Streamable HTTP transport may raise it when terminating sessions.
call_tool(name, arguments=None)
async
¶
Call a tool on the connected server.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Tool name. |
required |
arguments
|
dict[str, Any] | None
|
Tool arguments dict. |
None
|
Returns:
| Type | Description |
|---|---|
CallToolResult
|
MCP |
fetch_token(token_url, client_id, client_secret, *, timeout=10.0)
async
staticmethod
¶
Fetch a Bearer token from a server's token endpoint.
This is a convenience for development/testing when the MCP server
has a built-in token issuer enabled via
:meth:~promptise.mcp.server.MCPServer.enable_token_endpoint.
For production, obtain tokens from your Identity Provider
(Auth0, Keycloak, Okta, etc.) and pass them directly via
bearer_token.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token_url
|
str
|
Full URL of the token endpoint
(e.g. |
required |
client_id
|
str
|
Client identifier registered on the server. |
required |
client_secret
|
str
|
Client secret for authentication. |
required |
timeout
|
float
|
HTTP request timeout in seconds. |
10.0
|
Returns:
| Type | Description |
|---|---|
str
|
The access token string (ready to pass as |
Raises:
| Type | Description |
|---|---|
MCPClientError
|
On network errors or authentication failure. |
list_tools()
async
¶
List all tools from the connected server.
Returns:
| Type | Description |
|---|---|
list[Tool]
|
List of MCP |
MCPClientError¶
promptise.mcp.client.MCPClientError
¶
Bases: RuntimeError
Raised when an MCP client operation fails.
MCPMultiClient¶
promptise.mcp.client.MCPMultiClient
¶
Connect to multiple MCP servers and aggregate their tools.
Each server gets its own MCPClient with independent auth / headers.
Tools are tracked per-server so call_tool routes to the correct one.
Tool name collisions: If two servers expose a tool with the same name, the last-discovered server wins and a warning is logged. Consider using server-specific prefixes on your MCP servers to avoid collisions.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
clients
|
dict[str, MCPClient]
|
Mapping of server name → |
required |
Example::
multi = MCPMultiClient({
"hr": MCPClient(url="http://localhost:8080/mcp", bearer_token="..."),
"docs": MCPClient(url="http://localhost:9090/mcp", api_key="secret"),
})
async with multi:
tools = await multi.list_tools()
result = await multi.call_tool("search_employees", {"query": "python"})
servers
property
¶
Read-only view of server name → client mapping.
tool_to_server
property
¶
Read-only view of tool name → server name mapping.
__aenter__()
async
¶
Connect to all servers.
__aexit__(*exc)
async
¶
Disconnect from all servers.
call_tool(name, arguments=None)
async
¶
Call a tool, automatically routing to the correct server.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Tool name (as discovered via |
required |
arguments
|
dict[str, Any] | None
|
Tool arguments dict. |
None
|
Returns:
| Type | Description |
|---|---|
CallToolResult
|
MCP |
Raises:
| Type | Description |
|---|---|
MCPClientError
|
If the tool name is unknown or the call fails. |
list_tools()
async
¶
Discover tools from all connected servers.
Returns:
| Type | Description |
|---|---|
list[Tool]
|
Combined list of tools from all servers. The |
list[Tool]
|
|
list[Tool]
|
routes correctly. |
MCPToolAdapter¶
promptise.mcp.client.MCPToolAdapter
¶
Discover MCP tools and convert them to LangChain BaseTool instances.
Backed by the Promptise MCP Client.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
multi
|
MCPMultiClient
|
Connected |
required |
on_before
|
OnBefore | None
|
Callback fired before each tool invocation. |
None
|
on_after
|
OnAfter | None
|
Callback fired after each tool invocation. |
None
|
on_error
|
OnError | None
|
Callback fired on tool errors. |
None
|
Example::
multi = MCPMultiClient({"hr": MCPClient(...)})
async with multi:
adapter = MCPToolAdapter(multi)
tools = await adapter.as_langchain_tools()
# Pass `tools` to build_agent(extra_tools=tools)
as_langchain_tools()
async
¶
Discover tools and return them as LangChain BaseTool instances.
Each tool's args_schema is a recursively-built Pydantic model
that preserves nested object structure, descriptions, defaults,
and constraints from the MCP server's JSON Schema.
When optimize is set, static optimizations (schema
minification, description truncation) are applied to reduce
token cost.
Returns:
| Type | Description |
|---|---|
list[BaseTool]
|
List of |
list_tool_info()
async
¶
Return human-readable tool metadata for introspection.