dataflow.mcp.client¶

MCP Client Integration for haive-dataflow.

This module provides integration between MCP servers and the Haive framework through LangChain MCP adapters and the dataflow registry system.

The client handles: - Connection management to multiple MCP servers - Tool loading and registration from MCP servers - Integration with LangGraph workflows - Health monitoring and error handling

Classes:: MCPClient: Main client for MCP server integration MCPToolProvider: Provider for MCP tools in the registry MCPServerAdapter: Adapter for individual MCP servers

Classes¶

`MCPClient`	Client for managing MCP server connections and tools.
`MCPServerAdapter`	Adapter for individual MCP servers.
`MCPToolProvider`	Provider for registering MCP tools in the dataflow registry.

Module Contents¶

class dataflow.mcp.client.MCPClient(registry_system=None)¶

Client for managing MCP server connections and tools.

This class provides a high-level interface for connecting to MCP servers, loading tools, and integrating with the Haive dataflow registry system.

registry_system¶: Reference to the dataflow registry

mcp_client¶: Underlying MultiServerMCPClient instance

connected_servers¶: Dictionary of connected server configurations

available_tools¶: Cache of available tools from all servers

Examples

Basic usage:

from haive.dataflow.mcp import MCPClient from haive.dataflow import registry_system

client = MCPClient(registry_system) await client.initialize_from_registry()

# Get available tools tools = await client.get_available_tools()

# Execute a tool result = await client.execute_tool(“read_file”, {“path”: “/path/to/file”})

Initialize MCP client.

Parameters:: registry_system – Optional registry system instance

async check_server_health(server_name)¶

Check health of a specific MCP server.

Parameters:: server_name (str) – Name of the server to check
Returns:: Server health information
Return type:: haive.dataflow.registry.models.MCPServerHealth

async connect_to_servers(server_configs)¶

Connect to specific MCP servers.

Parameters:: server_configs (dict[str, haive.dataflow.registry.models.MCPServerConfig]) – Dictionary of server name to configuration
Returns:: True if connection successful, False otherwise
Return type:: bool

async execute_tool(tool_name, parameters)¶

Execute a specific MCP tool.

Parameters:

tool_name (str) – Name of the tool to execute
parameters (dict[str, Any]) – Tool parameters

Returns:

Tool execution result

Return type:

Any

async get_available_tools()¶

Get all available tools from connected MCP servers.

Returns:: List of LangChain Tool objects
Return type:: list[Any]

async get_server_health_status()¶

Get health status for all connected servers.

Returns:: Dictionary of server name to health status
Return type:: dict[str, haive.dataflow.registry.models.MCPServerHealth]

async initialize_from_registry()¶

Initialize MCP client with servers from the registry.

Returns:: True if initialization successful, False otherwise
Return type:: bool

class dataflow.mcp.client.MCPServerAdapter(config)¶

Adapter for individual MCP servers.

This class provides a consistent interface for working with individual MCP servers, handling connection, tool execution, and health monitoring.

Initialize MCP server adapter.

Parameters:: config (haive.dataflow.registry.models.MCPServerConfig) – Server configuration

async connect()¶

Connect to the MCP server.

Returns:: True if connection successful, False otherwise
Return type:: bool

async disconnect()¶: Disconnect from the MCP server.

async execute_tool(tool_name, parameters)¶

Execute a tool on this server.

Parameters:

tool_name (str) – Name of the tool to execute
parameters (dict[str, Any]) – Tool parameters

Returns:

Tool execution result

Return type:

Any

async get_available_tools()¶

Get list of available tools on this server.

Returns:: List of tool names
Return type:: list[str]

get_health_status()¶

Get current health status.

Returns:: Current health status
Return type:: haive.dataflow.registry.models.MCPServerHealth

class dataflow.mcp.client.MCPToolProvider(mcp_client, registry_system=None)¶

Provider for registering MCP tools in the dataflow registry.

This class handles the discovery and registration of tools from MCP servers into the Haive dataflow registry system for broader discovery and use.

Initialize MCP tool provider.

Parameters:

mcp_client (MCPClient) – MCP client instance
registry_system – Registry system for tool registration

async discover_and_register_tools()¶

Discover MCP tools and register them in the dataflow registry.

Returns:: List of registry IDs for registered tools
Return type:: list[str]