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

Attributes

DATAFLOW_AVAILABLE

logger

mcp

Classes

AgentCreationRequest

Request model for agent creation.

Functions

agent_configuration_prompt(→ list[dict[str, str]])

Generate a prompt for agent configuration.

component_search_prompt(→ list[dict[str, str]])

Generate a prompt for searching components by requirement.

create_agent(→ dict[str, Any])

Create a new Haive agent with specified configuration.

discover_components(→ dict[str, Any])

Discover Haive components in the system.

execute_tool(→ dict[str, Any])

Execute a Haive tool by name.

get_registry_entities(→ dict[str, Any])

Resource providing all registry entities.

get_registry_statistics(→ dict[str, Any])

Resource providing registry statistics.

initialize_server()

Initialize the MCP server with dataflow integration.

query_registry(→ list[dict[str, Any]])

Query the Haive registry for components.

Module Contents

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

Bases: pydantic.BaseModel

Request model for agent creation.

model: str = None
name: str = None
system_prompt: str | None = None
temperature: float = None
tools: list[str] = None
async mcp.servers.dataflow_mcp_server.agent_configuration_prompt(task_description: str, available_tools: list[str]) → list[dict[str, str]]

Generate a prompt for agent configuration.

Parameters:

  • task_description – Description of the task

  • available_tools – List of available tool names

Returns:

Prompt messages for agent configuration

async mcp.servers.dataflow_mcp_server.component_search_prompt(requirement: str) → list[dict[str, str]]

Generate a prompt for searching components by requirement.

Parameters:

requirement – User’s requirement description

Returns:

Prompt messages for component search

async mcp.servers.dataflow_mcp_server.create_agent(request: AgentCreationRequest) → dict[str, Any]

Create a new Haive agent with specified configuration.

Parameters:

request – Agent creation parameters

Returns:

Agent creation result with ID and status

async mcp.servers.dataflow_mcp_server.discover_components(component_type: str = 'all', auto_register: bool = False) → dict[str, Any]

Discover Haive components in the system.

Parameters:

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

  • auto_register – Whether to automatically register discovered components

Returns:

Discovery results with counts and component details

async mcp.servers.dataflow_mcp_server.execute_tool(tool_name: str, input_data: dict[str, Any]) → dict[str, Any]

Execute a Haive tool by name.

Parameters:

  • tool_name – Name of the tool to execute

  • input_data – Input parameters for the tool

Returns:

Tool execution result

async mcp.servers.dataflow_mcp_server.get_registry_entities() → dict[str, Any]

Resource providing all registry entities.

Returns:

Complete registry state

async mcp.servers.dataflow_mcp_server.get_registry_statistics() → dict[str, Any]

Resource providing registry statistics.

Returns:

Statistics about registered components

async mcp.servers.dataflow_mcp_server.initialize_server()

Initialize the MCP server with dataflow integration.

async mcp.servers.dataflow_mcp_server.query_registry(entity_type: str | None = None, name_pattern: str | None = None, limit: int = 10) → list[dict[str, Any]]

Query the Haive registry for components.

Parameters:

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

  • name_pattern – Pattern to match entity names

  • limit – Maximum number of results to return

Returns:

List of registry entries matching the query

mcp.servers.dataflow_mcp_server.DATAFLOW_AVAILABLE = True
mcp.servers.dataflow_mcp_server.logger
mcp.servers.dataflow_mcp_server.mcp