mcp.agents.transferable_mcp_agent

Transferable MCP agent implementation with resource/prompt/tool sharing capabilities.

This module provides an advanced MCP agent that supports sharing and transferring capabilities between agent instances. It enables collaborative workflows where multiple agents can share MCP clients, tools, resources, and prompts for efficient distributed processing.

The TransferableMCPAgent extends the basic MCPAgent with:

• Shared MCP client pools for resource efficiency

• Tool transfer mechanisms between agents

• Resource delegation and access control

• Prompt sharing for consistent behavior

• Collaborative session management

• Transfer tracking and auditing

Classes:

TransferableMCPAgent: Agent with MCP sharing and transfer capabilities

Functions:

create_collaborative_agents: Factory for creating collaborative agent groups

Example

Creating and using transferable agents:

from haive.mcp.agents import TransferableMCPAgent
from haive.mcp.config import MCPConfig

from haive import agents

# Create collaborative agents with shared client agents = TransferableMCPAgent.create_collaborative_agents(

engine=engine, mcp_config=mcp_config, num_agents=3, shared_client=True

)

# Initialize all agents for agent in agents:

await agent.setup()

# Transfer tools from first agent to others leader = agents[0] for follower in agents[1:]:

count = await leader.transfer_all_tools_to_agent(follower) print(f”Transferred {count} tools to {follower.name}”)

# Collaborate on a task results = await leader.collaborate_with_agents(

agents[1:], task=”Analyze repository structure”

)

Note

Shared clients reduce resource usage but require careful synchronization. Transfer operations are tracked for auditing and debugging purposes.

Classes

TransferableMCPAgent

An MCP-enabled agent with enhanced transferability features.

Module Contents

class mcp.agents.transferable_mcp_agent.TransferableMCPAgent(/, **data: Any)

Bases: haive.mcp.mixins.mcp_mixin.MCPMixin, haive.agents.simple.SimpleAgent

An MCP-enabled agent with enhanced transferability features.

TransferableMCPAgent extends SimpleAgent with sophisticated sharing and transfer mechanisms for MCP resources. It enables efficient multi-agent workflows by allowing agents to share clients, transfer tools, and collaborate on tasks.

The agent supports:

• Sharing MCP clients between agents to reduce resource usage

• Transferring resources/prompts/tools to other agents

• Session-based collaboration with shared state

• Dynamic capability delegation for flexible workflows

• Transfer tracking and auditing

share_client

Whether to share MCP client with other agents

client_pool_key

Key for shared client pool identification

_transferred_tools

Set of tool names transferred to other agents

_transferred_resources

Set of resource URIs delegated

_transferred_prompts

Set of prompts shared with other agents

Class Attributes:

_shared_mcp_clients: Pool of shared MCP client instances _shared_sessions: Pool of shared MCP sessions

Example

Basic usage with tool transfer:

# Create source agent with tools
source = TransferableMCPAgent(
    engine=engine,
    mcp_config=mcp_config,
    name="source_agent"
)
await source.setup()

# Create target agent
target = TransferableMCPAgent(
    engine=engine,
    mcp_config=minimal_config,
    name="target_agent"
)
await target.setup()

# Transfer specific tool
success = await source.transfer_tool_to_agent(
    target,
    "filesystem_read_file"
)

# Now target can use the transferred tool
result = await target.call_mcp_tool(
    "filesystem_read_file",
    {"path": "/path/to/file"}
)

async collaborate_with_agents(agents: list[TransferableMCPAgent], task: str) → dict[str, Any]

Collaborate with multiple agents on a task.

Orchestrates collaboration by sharing tools and resources with a group of agents for a specific task. This method sets up the collaborative environment but does not execute the task itself.

Parameters:

• agents – List of agents to collaborate with

• task – Description of the collaborative task

Returns:

Collaboration setup results including:

• task: The task description

• agents: List of agent details and tools received

• shared_tools: Total number of tools shared

• shared_resources: Total number of resources shared

Return type:

Dict[str, Any]

Example

Setting up a collaborative analysis:

classmethod create_collaborative_agents(engine: Any, mcp_config: haive.mcp.config.MCPConfig, num_agents: int = 2, shared_client: bool = True) → list[TransferableMCPAgent]

Create multiple collaborative agents with shared MCP client.

Factory method that creates a group of agents configured for collaboration. When shared_client is True, all agents will share the same MCP client instance for resource efficiency.

Parameters:

• engine – LLM engine configuration (AugLLMConfig)

• mcp_config – MCP configuration to use for all agents

• num_agents – Number of agents to create (default: 2)

• shared_client – Whether agents should share MCP client (default: True)

Returns:

List of configured collaborative agents

Return type:

List[TransferableMCPAgent]

Raises:

ValueError – If num_agents < 1

async delegate_resource_access(agent: TransferableMCPAgent, server_name: str, resource_uris: list[str] | None = None) → list[Any]

Delegate resource access to another agent.

Retrieves resources from an MCP server on behalf of another agent. This is useful when one agent has access to resources that another agent needs but cannot directly access.

Parameters:

• agent – Target agent requesting resource access

• server_name – Name of the MCP server providing resources

• resource_uris – Specific resource URIs to retrieve, or None for all

Returns:

Resources retrieved from the server

Return type:

List[Any]

Raises:

• RuntimeError – If MCP client not initialized

• ValueError – If server not found

Example

Delegating file access:

get_transfer_status() → dict[str, Any]

Get status of all transfers.

Returns a summary of all transfer operations performed by this agent, useful for auditing and debugging collaborative workflows.

Returns:

Transfer status including:

• transferred_tools: List of tool names transferred

• transferred_resources: List of resource URIs delegated

• transferred_prompts: List of prompts shared

• shared_client: Whether client sharing is enabled

• client_pool_key: Key used for client pooling

Return type:

Dict[str, Any]

Example

Checking transfer history:

async initialize_mcp() → bool

Initialize MCP with client sharing support.

This method extends the base initialization to support client sharing between agent instances. When share_client is True, it will attempt to reuse an existing client from the shared pool before creating a new one.

The client pool key is determined by:

1. Explicit client_pool_key if provided

2. Hash of the MCP configuration for automatic grouping

Returns:

True if initialization successful, False otherwise

Return type:

bool

Raises:

Exception – Logged but not raised, ensures graceful degradation

Example

Manual initialization with shared client:

agent1 = TransferableMCPAgent(
    engine=engine,
    mcp_config=config,
    share_client=True,
    client_pool_key="my_shared_pool"
)
await agent1.initialize_mcp()  # Creates new client

agent2 = TransferableMCPAgent(
    engine=engine,
    mcp_config=config,
    share_client=True,
    client_pool_key="my_shared_pool"
)
await agent2.initialize_mcp()  # Reuses agent1's client

async share_prompt_with_agent(agent: TransferableMCPAgent, server_name: str, prompt_name: str, arguments: dict[str, Any] | None = None) → list[Any]

Share a prompt with another agent.

Retrieves a prompt from an MCP server and shares it with another agent. This enables consistent prompting across multiple agents for coordinated behavior.

Parameters:

• agent – Target agent to receive the prompt

• server_name – Name of the MCP server providing the prompt

• prompt_name – Name of the prompt to retrieve

• arguments – Optional arguments to customize the prompt

Returns:

Prompt messages that can be used by the target agent

Return type:

List[Any]

Raises:

• RuntimeError – If MCP client not initialized

• ValueError – If server or prompt not found

Example

Sharing a code review prompt:

async shared_mcp_session(session_key: str, server_name: str | None = None)

Create or join a shared MCP session.

Context manager that provides shared MCP sessions for coordinated operations between multiple agents. Sessions are identified by unique keys and automatically cleaned up when all agents exit.

Parameters:

• session_key – Unique identifier for the shared session

• server_name – Optional specific server for the session

Yields:

Any – Shared MCP session instance

Example

Coordinated file operations:

async transfer_all_tools_to_agent(agent: TransferableMCPAgent) → int

Transfer all tools to another agent.

Bulk transfer operation that shares all available MCP tools from this agent to the target agent. Useful for creating worker agents with full capabilities.

Parameters:

agent – Target agent to receive all tools

Returns:

Number of tools successfully transferred

Return type:

int

Example

Creating a fully-equipped worker:

# Leader has all MCP servers configured
leader = TransferableMCPAgent(
    engine=engine,
    mcp_config=full_config,
    name="leader"
)
await leader.setup()

# Worker starts with minimal config
worker = TransferableMCPAgent(
    engine=engine,
    mcp_config=minimal_config,
    name="worker"
)
await worker.setup()

# Transfer all tools to worker
count = await leader.transfer_all_tools_to_agent(worker)
print(f"Worker now has {count} additional tools")

async transfer_tool_to_agent(agent: TransferableMCPAgent, tool_name: str) → bool

Transfer a specific tool to another agent.

Transfers a single MCP tool from this agent to another agent. The tool reference is shared, not copied, so both agents will have access to the same tool instance. The transfer is tracked for auditing.

Parameters:

• agent – Target agent to receive the tool

• tool_name – Name of the tool to transfer (e.g., “filesystem_read_file”)

Returns:

True if transfer successful, False if tool not found

Return type:

bool

Raises:

RuntimeError – If target agent initialization fails

Example

Transferring a specific tool:

# Transfer GitHub issue creation tool
success = await source_agent.transfer_tool_to_agent(
    target_agent,
    "github_create_issue"
)

if success:
    # Target can now create GitHub issues
    await target_agent.call_mcp_tool(
        "github_create_issue",
        {"title": "New feature", "body": "Description"}
    )

client_pool_key: str | None = None

share_client: bool = None