MCPElement

The purpose of this Element is to enable agentic capabilities when coupled with LLMs. It facilitates tool invocation via the Model Context Protocol (MCP). It can start and manage MCP servers (e.g., script-based or SSE servers), wrap Python functions as tools, and expose Pydantic schemas for tool arguments. When receiving a StructuredPayload containing the tool names and arguments, it emits a ToolResponsePayload containing the callbacks for the chosen tools to run, as well other metadata related to the tool call. Because it expects structured input, it also has the capacity to send out a pydantic schema containing the exact format it expects, along with the tool names and expected arguments.

Instantiation

Arguments:

mcps: dict
A mapping of server names to MCP configurations. Each configuration dict may include the following keys: type: str
Protocol type; one of ‘script’, ‘sse’, ‘mcp_class’, or ‘functions’. script: str (optional)
Path to the script for ‘script’ type servers. command: str (optional)
Executable to run the script (defaults to the Python interpreter). args: list[str] (optional)
Command-line arguments for the script. env: dict (optional)
Environment variables for the subprocess. host: str (optional)
Host address for ‘sse’ type servers. port: int (optional)
Port number for ‘sse’ type servers. tools: dict[str, Callable] (optional)
Mapping of function names to Python callables for ‘functions’ type. tools_requiring_permission: list[str] (optional)
List of tool names that require user permission.

Example:

mcps = {
    'todo': {
        'type': 'script',
        'script': 'todo_server.py',
        'command': 'python',
        'args': ['--logging'],
        'env': {'API_KEY': 'xyz'},
        'tools_requiring_permission': ['remove_todo']
    },
    'weather': {
        'type': 'sse',
        'host': 'localhost',
        'port': 1234
    },
    'custom_funcs': {
        'type': 'functions',
        'tools': {
            'calculate': calculate,
            'get_current_time': get_current_time
        },
        'tools_requiring_permission': ['calculate']
    }
}

loop: Any, optional
The asyncio event loop to use for setup. Defaults to the main loop from LoopRegistry.

Input Ports

Port Name	Payload Type	Behavior
tool_request_structured_input	StructuredPayload	Receives a Payload whose model.data is a list of dicts, each containing: name: str — hybrid tool name. parameters: dict — tool arguments (optional). Example Unpacks and forwards these entries to tools_response_output.

Output Ports

Port Name	Payload Type	Behavior
tools_schema_output	SchemaPayload	Emits a SchemaPayload containing the Pydantic schema for available tools.
tools_response_output	ToolsResponsePayload	Emits a ToolsResponsePayload containing tool results mapping tool names to their responses. Example

Data Flow Details

tool_request_structured_input payload

[
    {'name': 'todo_add', 'parameters': {'task': 'Buy milk'}},
    {'name': 'weather_get', 'parameters': {'city': 'SF'}}
]

tools_response_output payload

{
    'weather_temp_mcp': {
        'mcp_name': 'weather_mcp',
        'tool_name': 'temp',
        'permission_required': False,
        'description': 'Get the temperature in a location',
        'parameters': {'location': 'San Francisco'},
        'response': {
            'meta': None,
            'content': [{
                'type': 'text',
                'text': 'The temperature is 54F.',
                'annotations': None
            }],
            'isError': False
        }
    },
    'custom_funcs_check_prime': {
        'mcp_name': 'custom_funcs',
        'tool_name': 'check_prime',
        'permission_required': False,
        'description': 'Check if a number is prime.',
        'parameters': {'n': 17},
        'response': {
            'meta': None,
            'content': [{
                'type': 'text',
                'text': 'True',
                'annotations': None
            }],
            'isError': False
        }
    }
}