Skip to main content

Overview

The McpAgent extends ToolAgent with the ability to connect to MCP (Model Context Protocol) servers. It provides access to tools, resources, and prompts from external servers, enabling rich integrations with external systems and data sources. McpAgent uses an MCPAggregator to manage multiple server connections and provides a unified interface for accessing capabilities across all connected servers.

Key Features

  • MCP Server Integration: Connect to multiple MCP servers simultaneously
  • Dynamic Server Management: Attach and detach servers at runtime
  • Resource Access: Read resources from MCP servers (files, databases, APIs)
  • Prompt Templates: Use server-provided prompt templates
  • Tool Aggregation: Access tools from all connected servers
  • Connection Persistence: Maintain server connections across requests
  • Skill Support: Load and use skills with filesystem access
  • Shell Runtime: Optional local shell execution capabilities

Architecture

McpAgent
    ↓ extends
ToolAgent (function tools + MCP tools)
    ↓ extends  
LlmAgent (conversation & display)
    ↓ uses
MCPAggregator (server connection management)

Creating an MCP Agent

Basic Configuration

import asyncio
from fast_agent.agents.agent_types import AgentConfig
from fast_agent.agents.mcp_agent import McpAgent
from fast_agent.core import Core
from fast_agent.llm.model_factory import ModelFactory

async def main():
    core = Core()
    await core.initialize()
    
    # Create agent configuration with MCP servers
    config = AgentConfig(
        name="research_bot",
        instruction="You are a research assistant with access to web search.",
        model="gpt-4o-mini",
        servers=["fetch"]  # MCP server names from config
    )
    
    # Create the MCP agent
    agent = McpAgent(
        config,
        connection_persistence=True,
        context=core.context
    )
    
    # Attach LLM and initialize MCP connections
    await agent.attach_llm(ModelFactory.create_factory("gpt-4o-mini"))
    await agent.initialize()
    
    # Use MCP tools
    result = await agent.send(
        "Find information about Fast Agent at https://fast-agent.ai/"
    )
    print(result)
    
    await agent.shutdown()
    await core.cleanup()

asyncio.run(main())

Multiple MCP Servers

config = AgentConfig(
    name="multi_server_agent",
    instruction="You have access to web search, filesystem, and database tools.",
    servers=[
        "fetch",      # Web fetching
        "filesystem", # File operations  
        "postgres"    # Database access
    ],
    model="claude-3-5-sonnet-20241022"
)

agent = McpAgent(config, context=core.context)
await agent.attach_llm(ModelFactory.create_factory("claude-3-5-sonnet-20241022"))
await agent.initialize()

Configuration Options

AgentConfig for MCP

servers
list[str]
default:"[]"
List of MCP server names to connect to (from settings.json)
tools
dict[str, list[str]]
default:"{}"
Filter which tools to expose from each server
tools={
    "fetch": ["fetch_url", "search_*"],  # Only these tools
    "filesystem": ["*"]  # All tools
}
resources
dict[str, list[str]]
default:"{}"
Filter which resources to expose from each server
prompts
dict[str, list[str]]
default:"{}"
Filter which prompts to expose from each server
skills
SkillConfig
Skills configuration (enables shell runtime automatically)

Constructor Parameters

connection_persistence
bool
default:"true"
Whether to maintain server connections across requests
context
Context
Fast Agent context (required for server configuration)

Working with MCP Tools

Listing Available Tools

# List all tools from all servers
tools_result = await agent.list_tools()

for tool in tools_result.tools:
    print(f"Tool: {tool.name}")
    print(f"Description: {tool.description}")
    print(f"Schema: {tool.inputSchema}")

Calling Tools Directly

from mcp.types import CallToolResult

# Call a specific MCP tool
result: CallToolResult = await agent.call_tool(
    name="fetch__fetch_url",  # Namespaced tool name
    arguments={"url": "https://example.com"},
)

if not result.isError:
    print(result.content[0].text)
else:
    print(f"Error: {result.content[0].text}")

Working with Resources

Listing Resources

# List all resources from all servers
resources = await agent.list_resources()

for server_name, resource_list in resources.items():
    print(f"Server: {server_name}")
    for resource_uri in resource_list:
        print(f"  - {resource_uri}")

Reading Resources

from mcp.types import ReadResourceResult

# Read a resource
result: ReadResourceResult = await agent.get_resource(
    resource_uri="file:///path/to/file.txt",
    server_name="filesystem"  # Optional: specify server
)

for content in result.contents:
    if hasattr(content, 'text'):
        print(content.text)

Using Resources in Prompts

# Attach resource to a message
response = await agent.with_resource(
    prompt_content="Analyze this file",
    resource_uri="file:///data/report.csv",
    server_name="filesystem"
)
print(response)

# Or get embedded resources for custom use
embedded = await agent.get_embedded_resources(
    resource_uri="file:///config.json"
)

Working with Prompts

Applying MCP Prompts

# Apply a prompt template from an MCP server
response = await agent.apply_prompt(
    prompt="code_review",
    arguments={"language": "python", "file": "agent.py"},
    namespace="dev_tools"  # Server name
)
print(response)

# Use as persistent template
await agent.apply_prompt(
    prompt="system_context",
    as_template=True  # Always included in context
)

Getting Prompt Details

from mcp.types import GetPromptResult

# Get prompt metadata
prompt_result: GetPromptResult = await agent.get_prompt(
    prompt_name="code_review",
    arguments={"language": "python"},
    server_name="dev_tools"
)

print(f"Description: {prompt_result.description}")
for msg in prompt_result.messages:
    print(f"{msg.role}: {msg.content}")

Dynamic Server Management

Attaching Servers at Runtime

from fast_agent.mcp.mcp_aggregator import MCPAttachOptions

# Attach a new server
result = await agent.attach_mcp_server(
    server_name="new_server",
    server_config=None,  # Uses config from settings
    options=MCPAttachOptions(
        timeout_seconds=30,
        reconnect=True
    )
)

if result.success:
    print(f"Connected to {result.server_name}")
    print(f"Tools: {len(result.tools)}")
else:
    print(f"Failed: {result.error}")

Detaching Servers

# Disconnect a server
result = await agent.detach_mcp_server("new_server")

if result.success:
    print(f"Disconnected from {result.server_name}")

Listing Connected Servers

# Get list of attached servers
attached = agent.list_attached_mcp_servers()
print(f"Connected servers: {attached}")

# Get detailed server status
status = await agent.get_server_status()
for server_name, server_status in status.items():
    print(f"{server_name}: {server_status.state}")
    if server_status.error:
        print(f"  Error: {server_status.error}")

Shell Runtime

Enabling Shell Access

from pathlib import Path

# Enable shell runtime after creation
agent.enable_shell(working_directory=Path("/path/to/project"))

# Check if enabled
if agent.shell_runtime_enabled:
    print(f"Shell access modes: {agent.shell_access_modes}")

Using Shell Tools

# The agent automatically gets bash/shell tools when enabled
result = await agent.send(
    "List all Python files in the current directory"
)
# Agent can now execute: ls *.py

Skills Integration

Configuring Skills

from fast_agent.skills import SkillManifest

config = AgentConfig(
    name="skilled_agent",
    skills=["code-review", "testing"],  # Skill names
    model="gpt-4o"
)

agent = McpAgent(config, context=core.context)
await agent.attach_llm(ModelFactory.create_factory("gpt-4o"))
await agent.initialize()

# Skills are automatically available via read_skill tool

Accessing Skill Manifests

# Get configured skills
manifests = agent.skill_manifests

for manifest in manifests:
    print(f"Skill: {manifest.name}")
    print(f"Path: {manifest.path}")
    print(f"Description: {manifest.description}")

Filtering Tools and Resources

Pattern-Based Filtering

config = AgentConfig(
    name="filtered_agent",
    servers=["filesystem", "web"],
    tools={
        "filesystem": ["read_*", "list_*"],  # Read-only tools
        "web": ["fetch_url"]  # Only URL fetching
    },
    resources={
        "filesystem": ["file:///data/*"]  # Only /data files
    }
)

Dynamic Filtering

Filtering is applied automatically based on the configuration. The agent only exposes matching tools and resources to the LLM.

Advanced Features

Instruction Templates

# Use server instructions in system prompt
config = AgentConfig(
    name="template_agent",
    instruction="""You are a helpful assistant.
    
    {{serverInstructions}}
    
    {{agentSkills}}
    """,
    servers=["fetch"],
    skills=["research"]
)

# Templates are resolved during initialization
agent = McpAgent(config, context=core.context)
await agent.attach_llm(ModelFactory.create_factory("gpt-4o-mini"))
await agent.initialize()

# View resolved instruction
print(agent.instruction)

Custom Filesystem Runtime

from fast_agent.tools.local_filesystem_runtime import LocalFilesystemRuntime

# Create custom filesystem runtime
runtime = LocalFilesystemRuntime(
    logger=agent.logger,
    working_directory=Path("/custom/path"),
    enable_read=True,
    enable_write=True,
    enable_apply_patch=False
)

# Inject into agent
agent.set_filesystem_runtime(runtime)

Best Practices

  • Configure servers in settings.json for reusability
  • Use connection_persistence=True for better performance
  • Implement retry logic for transient connection failures
  • Monitor server status for health checks
  • Use filters to limit exposed capabilities
  • Combine multiple servers for specialized agents
  • Test tool combinations thoroughly
  • Document which tools are available to users
  • Cache frequently accessed resources
  • Use resource templates for dynamic URIs
  • Validate resource URIs before access
  • Handle missing resources gracefully
  • Check server connection status before requests
  • Handle tool execution errors explicitly
  • Implement fallback behavior for missing servers
  • Log server errors for debugging

Common Patterns

Research Agent

config = AgentConfig(
    name="researcher",
    instruction="You are a research assistant that can search the web and read files.",
    servers=["fetch", "filesystem"],
    tools={
        "fetch": ["fetch_url", "search_*"],
        "filesystem": ["read_*"]
    },
    model="gpt-4o"
)

Data Analysis Agent

config = AgentConfig(
    name="analyst",
    instruction="You analyze data from various sources.",
    servers=["postgres", "filesystem"],
    model="claude-3-5-sonnet-20241022"
)

Development Agent

config = AgentConfig(
    name="dev_assistant",
    instruction="You help with software development tasks.",
    servers=["filesystem"],
    skills=["code-review", "testing"],
    shell=True,
    model="gpt-4o"
)

Next Steps

MCP Servers

Learn about available MCP servers and creating custom ones

Tool Agent

Understand the underlying tool execution capabilities

Skills

Create and use agent skills

Resources

Work with MCP resources in detail