haive.mcp.servers.dataflow_mcp_server

MCP Server integrating Haive Dataflow for dynamic tool creation.

This server demonstrates how to create MCP tools from Haive’s dataflow system, exposing dataflow graphs as MCP tools and resources. It integrates with: - haive-core: For agent and engine functionality - haive-dataflow: For registry and discovery systems - haive-tools: For exposing existing tools via MCP

Classes

AgentCreationRequest

Request model for agent creation.

Functions

agent_configuration_prompt(task_description, ...)	Generate a prompt for agent configuration.
component_search_prompt(requirement)	Generate a prompt for searching components by requirement.
create_agent(request)	Create a new Haive agent with specified configuration.
discover_components([component_type, auto_register])	Discover Haive components in the system.
execute_tool(tool_name, input_data)	Execute a Haive tool by name.
get_registry_entities()	Resource providing all registry entities.
get_registry_statistics()	Resource providing registry statistics.
initialize_server()	Initialize the MCP server with dataflow integration.
query_registry([entity_type, name_pattern, limit])	Query the Haive registry for components.

Module Contents

class haive.mcp.servers.dataflow_mcp_server.AgentCreationRequest(/, **data)

Bases: pydantic.BaseModel

Request model for agent creation.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

async haive.mcp.servers.dataflow_mcp_server.agent_configuration_prompt(task_description, available_tools)

Generate a prompt for agent configuration.

Parameters:

• task_description (str) – Description of the task

• available_tools (list[str]) – List of available tool names

Returns:

Prompt messages for agent configuration

Return type:

list[dict[str, str]]

async haive.mcp.servers.dataflow_mcp_server.component_search_prompt(requirement)

Generate a prompt for searching components by requirement.

Parameters:

requirement (str) – User’s requirement description

Returns:

Prompt messages for component search

Return type:

list[dict[str, str]]

async haive.mcp.servers.dataflow_mcp_server.create_agent(request)

Create a new Haive agent with specified configuration.

Parameters:

request (AgentCreationRequest) – Agent creation parameters

Returns:

Agent creation result with ID and status

Return type:

dict[str, Any]

async haive.mcp.servers.dataflow_mcp_server.discover_components(component_type='all', auto_register=False)

Discover Haive components in the system.

Parameters:

• component_type (str) – Type to discover (agents, tools, or all)

• auto_register (bool) – Whether to automatically register discovered components

Returns:

Discovery results with counts and component details

Return type:

dict[str, Any]

async haive.mcp.servers.dataflow_mcp_server.execute_tool(tool_name, input_data)

Execute a Haive tool by name.

Parameters:

• tool_name (str) – Name of the tool to execute

• input_data (dict[str, Any]) – Input parameters for the tool

Returns:

Tool execution result

Return type:

dict[str, Any]

async haive.mcp.servers.dataflow_mcp_server.get_registry_entities()

Resource providing all registry entities.

Returns:

Complete registry state

Return type:

dict[str, Any]

async haive.mcp.servers.dataflow_mcp_server.get_registry_statistics()

Resource providing registry statistics.

Returns:

Statistics about registered components

Return type:

dict[str, Any]

async haive.mcp.servers.dataflow_mcp_server.initialize_server()

Initialize the MCP server with dataflow integration.

async haive.mcp.servers.dataflow_mcp_server.query_registry(entity_type=None, name_pattern=None, limit=10)

Query the Haive registry for components.

Parameters:

• entity_type (str | None) – Type of entity to query (agent, tool, engine, etc.)

• name_pattern (str | None) – Pattern to match entity names

• limit (int) – Maximum number of results to return

Returns:

List of registry entries matching the query

Return type:

list[dict[str, Any]]