MCP Module

Access external tools and resources via MCP servers configured on the host.

Overview

AgentVisor provides two MCP modules:

Module	Use Case
langchain	LangChain/LangGraph integration (recommended)
mcp	Low-level API for direct control

All MCP operations are policy-checked by the host runtime. See MCP Integration for architecture details.

LangChain Integration

The recommended way to use MCP tools is via the LangChain adapter:

from agentvisor.langchain import get_mcp_tools

get_mcp_tools()

def get_mcp_tools(server_name: str | None = None) -> list[StructuredTool]:
    """Convert MCP tools to LangChain StructuredTool instances.

    Args:
        server_name: Optional filter by MCP server name. None = all servers.

    Returns:
        List of LangChain StructuredTool instances. Returns an empty list
        if no MCP servers are configured on the host.
    """

Example: ReAct Agent

from agentvisor.langchain import get_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_ollama import ChatOllama

# Get all MCP tools as LangChain tools
tools = get_mcp_tools()

# Create agent with MCP tools
model = ChatOllama(model="llama3.2")
graph = create_react_agent(model, tools)

Example: Filter by Server

# Only filesystem tools
fs_tools = get_mcp_tools(server_name="filesystem")

# Only GitHub tools
github_tools = get_mcp_tools(server_name="github")

# Combine specific servers
tools = fs_tools + github_tools

Tool Naming

Tools are named {server}__{tool} to avoid conflicts:

tools = get_mcp_tools()
for tool in tools:
    print(tool.name)
# filesystem__read_file
# filesystem__write_file
# github__get_repo
# github__list_issues

Low-Level API

For direct control over MCP operations:

from agentvisor import mcp

get_tools()

Framework-agnostic equivalent of agentvisor.langchain.get_mcp_tools() — returns MCPTool instances instead of LangChain StructuredTools, for use with any framework adapter:

def get_tools(server_name: str | None = None) -> list[MCPTool]:
    """Get MCP tools as framework-agnostic MCPTool instances.

    Args:
        server_name: Optional filter by MCP server name. None = all servers.

    Returns:
        List of MCPTool instances. Returns an empty list if no MCP servers
        are configured on the host.
    """

Each MCPTool has name ("<server>__<tool>"), description, input_schema (JSON Schema), and an invoke callable that takes a dict of arguments and returns the tool result as a string.

list_tools()

def list_tools(server_name: str | None = None) -> list[dict[str, Any]]:
    """List available MCP tools.

    Args:
        server_name: Optional filter by server name. None/empty = all servers.

    Returns:
        List of tool definitions. Each has:
            - server_name: str
            - tool_name: str
            - description: str
            - input_schema: dict (JSON Schema)

        Returns an empty list if no MCP servers are configured on the host.
    """

Example

from agentvisor import mcp

# List all tools
tools = mcp.list_tools()
for tool in tools:
    print(f"{tool['server_name']}/{tool['tool_name']}: {tool['description']}")

# List tools from specific server
fs_tools = mcp.list_tools(server_name="filesystem")

call_tool()

def call_tool(
    server_name: str,
    tool_name: str,
    arguments: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Call an MCP tool.

    Args:
        server_name: The MCP server name.
        tool_name: The tool to invoke.
        arguments: Tool arguments (key-value pairs).

    Returns:
        Dict with:
            - content: list of content items (each has type, text/data/uri)
            - is_error: bool
            - error_message: str (if is_error)
            - allowed: bool (False if denied by policy)
            - denial_reason: str (if not allowed)

    Raises:
        RuntimeError: If MCP is not configured on the host.
    """

Example

from agentvisor import mcp

result = mcp.call_tool(
    server_name="filesystem",
    tool_name="read_file",
    arguments={"path": "/data/config.json"}
)

if not result.get("allowed", True):
    print(f"Denied: {result['denial_reason']}")
elif result.get("is_error"):
    print(f"Error: {result['error_message']}")
else:
    for content in result["content"]:
        if content.get("text"):
            print(content["text"])

list_resources()

def list_resources(server_name: str | None = None) -> list[dict[str, Any]]:
    """List available MCP resources.

    Args:
        server_name: Optional filter by server name. None/empty = all servers.

    Returns:
        List of resource definitions. Each has:
            - server_name: str
            - uri: str
            - name: str
            - description: str
            - mime_type: str

        Returns an empty list if no MCP servers are configured on the host.
    """

Example

from agentvisor import mcp

resources = mcp.list_resources()
for r in resources:
    print(f"{r['uri']} ({r['mime_type']})")

read_resource()

def read_resource(server_name: str, uri: str) -> dict[str, Any]:
    """Read an MCP resource.

    Args:
        server_name: The MCP server name.
        uri: The resource URI.

    Returns:
        Dict with:
            - contents: list of content items
            - allowed: bool (False if denied by policy)
            - denial_reason: str (if not allowed)

    Raises:
        RuntimeError: If MCP is not configured on the host.
    """

Example

from agentvisor import mcp

result = mcp.read_resource("filesystem", "file:///data/readme.md")

if result.get("allowed", True):
    for content in result["contents"]:
        print(content.get("text", ""))

Response Handling

Policy Denials

When a policy denies an MCP operation, the response includes allowed: false instead of raising an exception:

result = mcp.call_tool("github", "delete_repo", {"repo": "important"})

if not result.get("allowed", True):
    # Handle denial gracefully
    print(f"Access denied: {result['denial_reason']}")
    # Example: "Policy denied access to mrn:agentvisor:mcp:github/delete_repo"

The LangChain adapter handles this automatically by returning a denial message:

# With get_mcp_tools(), denied calls return a message instead of failing
tools = get_mcp_tools()
result = tools[0].invoke({"repo": "important"})
# Returns: "Tool call denied: Policy denied access to ..."

Tool Errors

Tool errors from the MCP server are returned in the response:

result = mcp.call_tool("filesystem", "read_file", {"path": "/nonexistent"})

if result.get("is_error"):
    print(f"Tool error: {result['error_message']}")
    # Example: "File not found: /nonexistent"

Content Types

MCP tools can return multiple content types:

result = mcp.call_tool("server", "tool", {})

for content in result.get("content", []):
    if content.get("type") == "text":
        print(content["text"])
    elif content.get("type") == "image":
        # Base64-encoded image data
        image_data = content["data"]
    elif content.get("type") == "resource":
        # URI reference to another resource
        uri = content["uri"]

Error Handling

list_tools(), list_resources(), and get_tools()/get_mcp_tools() never raise for an unconfigured host — they return an empty list. Only call_tool() and read_resource() raise RuntimeError when MCP is not configured:

from agentvisor import mcp

try:
    result = mcp.call_tool("filesystem", "read_file", {"path": "/data/readme.md"})
except RuntimeError as e:
    # MCP not configured on host
    print(f"MCP unavailable: {e}")

Environment

The MCP module requires these environment variables (set automatically by AgentVisor):

Variable	Description
AGENTVISOR_AGENT_SOCKET	Path to agent Unix socket
AGENTVISOR_THREAD_ID	Current thread ID

Related Documentation

• MCP Gateway Guide - Architecture, security model, and configuration
• MCP Transport Guide - Exposing agents as MCP tools
• Configuration Reference - MCP server configuration