MCP Tools

Overview

MCP tools are executable functions that the server exposes to clients. Each tool has a defined schema, inputs, and outputs, enabling AI agents to perform specific actions.

Tools enable AI agents to interact with external systems, execute code, search data, and perform complex operations beyond text generation.

Available Tools

Agent Chat Tool

Send a message to the AI agent and receive an intelligent response optimized for agent consumption.

string

default:"agent_chat"

Tool identifier (namespaced for clarity)

Backward Compatibility: The old tool name chat is still supported for backward compatibility.

Input Schema:

{
  "type": "object",
  "properties": {
    "message": {
      "type": "string",
      "description": "The message or question to send to the agent",
      "minLength": 1,
      "maxLength": 10000
    },
    "username": {
      "type": "string",
      "description": "Username for authentication"
    },
    "thread_id": {
      "type": "string",
      "description": "Optional conversation ID to maintain context (e.g., 'conv_123')",
      "pattern": "^conv_[a-zA-Z0-9]+$"
    },
    "response_format": {
      "type": "string",
      "enum": ["concise", "detailed"],
      "default": "concise",
      "description": "Response verbosity level. 'concise' returns ~500 tokens (2-5 sec), 'detailed' returns ~2000 tokens (5-10 sec)"
    }
  },
  "required": ["message", "username"]
}

Response Format Control: New in v2.6.0. Agents can optimize for speed vs comprehensiveness:

concise (default): ~500 tokens, 2-5 seconds - ideal for quick answers
detailed: ~2000 tokens, 5-10 seconds - comprehensive explanations

Follows Anthropic’s best practice: “Expose a response_format enum parameter for token efficiency”

Example Usage:

## Basic usage with default concise format
response = await client.call_tool("agent_chat", {
    "message": "Explain quantum entanglement",
    "username": "alice",
    "thread_id": "conv_123"
})

print(response.content[0].text)

## Request detailed response for comprehensive answer
detailed_response = await client.call_tool("agent_chat", {
    "message": "Explain quantum computing in detail",
    "username": "alice",
    "response_format": "detailed"  # Get comprehensive response
})

## Backward compatible - old tool name still works
legacy_response = await client.call_tool("chat", {
    "message": "Hello",
    "username": "alice"
})

Response:

{
  "content": [
    {
      "type": "text",
      "text": "Quantum entanglement is a phenomenon where two or more particles become correlated in such a way that the quantum state of each particle cannot be described independently..."
    }
  ],
  "isError": false,
  "metadata": {
    "model": "claude-sonnet-4-5-20250929",
    "conversation_id": "conv_123",
    "usage": {
      "prompt_tokens": 20,
      "completion_tokens": 150,
      "total_tokens": 170
    }
  }
}

Conversation Search Tool

Search conversations using keywords and filters. Replaces the old list_conversations tool following Anthropic’s guidance: “Implement search-focused tools rather than list-all tools.”

string

default:"conversation_search"

Tool identifier (namespaced for clarity)

Breaking Change (v2.6.0): The list_conversations tool has been replaced with conversation_search to prevent context overflow. Backward-compatible routing still supports the old name.

Input Schema:

{
  "type": "object",
  "properties": {
    "query": {
      "type": "string",
      "description": "Search query to filter conversations",
      "minLength": 1,
      "maxLength": 500
    },
    "username": {
      "type": "string",
      "description": "Username for authentication"
    },
    "limit": {
      "type": "integer",
      "description": "Maximum number of conversations to return (1-50)",
      "minimum": 1,
      "maximum": 50,
      "default": 10
    }
  },
  "required": ["query", "username"]
}

Why Search-Focused? Following Anthropic’s best practices:

Prevents context overflow: Listing all conversations can consume thousands of tokens
Forces specificity: Agents must be explicit about what they’re looking for
More efficient: Returns only relevant results, not everything
Scalable: Works with users who have hundreds of conversations

Benefits over list-all:

Up to 50x reduction in response tokens for users with many conversations
Helpful truncation messages when results exceed limit
Sorted by relevance to query

Example Usage:

## Search for specific conversations
results = await client.call_tool("conversation_search", {
    "query": "project alpha",
    "username": "alice",
    "limit": 10
})

print(results.content[0].text)
## Output: Found 2 conversation(s) matching 'project alpha':
## 1. conversation:project_alpha_planning
## 2. conversation:project_alpha_review

## Empty query returns recent conversations
recent = await client.call_tool("conversation_search", {
    "query": "",
    "username": "alice",
    "limit": 5
})

Response:

{
  "content": [
    {
      "type": "text",
      "text": "Found 2 conversation(s) matching 'project alpha':\n1. conversation:project_alpha_planning\n2. conversation:project_alpha_review"
    }
  ],
  "isError": false
}

Truncation Guidance: When results exceed the limit:

{
  "content": [
    {
      "type": "text",
      "text": "Found 15 conversation(s) matching 'meeting':\n1. conversation:weekly_meeting\n...\n10. conversation:team_sync\n\n[Showing 10 of 15 results. Use a more specific query to narrow results.]"
    }
  ]
}

Conversation Get Tool

Retrieve a specific conversation thread by ID.

string

default:"conversation_get"

Tool identifier (namespaced for clarity)

Best Practice: Use conversation_search first to find conversation IDs, then use conversation_get to retrieve specific conversations.

Input Schema:

{
  "type": "object",
  "properties": {
    "thread_id": {
      "type": "string",
      "description": "Conversation thread identifier (e.g., 'conv_abc123')"
    },
    "username": {
      "type": "string",
      "description": "Username for authentication"
    }
  },
  "required": ["thread_id", "username"]
}

Example Usage:

## First, search for conversations
search_results = await client.call_tool("conversation_search", {
    "query": "project updates",
    "username": "alice"
})

## Then, get a specific conversation
conversation = await client.call_tool("conversation_get", {
    "thread_id": "conv_abc123",
    "username": "alice"
})

print(conversation.content[0].text)

Response:

{
  "content": [
    {
      "type": "text",
      "text": "Conversation history for thread conv_abc123"
    }
  ],
  "isError": false
}

Execute Code Tool

Execute code in a sandboxed environment (if enabled).

string

default:"execute_code"

Tool identifier

Input Schema:

{
  "type": "object",
  "properties": {
    "code": {
      "type": "string",
      "description": "Code to execute",
      "minLength": 1,
      "maxLength": 50000
    },
    "language": {
      "type": "string",
      "description": "Programming language",
      "enum": ["python", "javascript", "bash"],
      "default": "python"
    },
    "timeout": {
      "type": "integer",
      "description": "Execution timeout in seconds",
      "minimum": 1,
      "maximum": 30,
      "default": 10
    }
  },
  "required": ["code"]
}

Example Usage:

result = await client.call_tool("execute_code", {
    "code": """
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

print(fibonacci(10))
""",
    "language": "python",
    "timeout": 5
})

print(result.content[0].text)
## Output: 55

Response:

{
  "content": [
    {
      "type": "text",
      "text": "55\n"
    }
  ],
  "isError": false,
  "metadata": {
    "execution_time": 0.023,
    "language": "python"
  }
}

Database Query Tool

Query databases (if configured and authorized).

string

default:"query_database"

Tool identifier

Input Schema:

{
  "type": "object",
  "properties": {
    "query": {
      "type": "string",
      "description": "SQL query (SELECT only)",
      "pattern": "^SELECT.*"
    },
    "database": {
      "type": "string",
      "description": "Database name",
      "enum": ["analytics", "users", "products"]
    },
    "limit": {
      "type": "integer",
      "description": "Maximum rows to return",
      "minimum": 1,
      "maximum": 1000,
      "default": 100
    }
  },
  "required": ["query", "database"]
}

Example Usage:

results = await client.call_tool("query_database", {
    "query": "SELECT name, email FROM users WHERE active = true",
    "database": "users",
    "limit": 10
})

Response:

{
  "content": [
    {
      "type": "text",
      "text": "| name | email |\n|------|-------|\n| Alice | alice@example.com |\n| Bob | bob@example.com |"
    }
  ],
  "isError": false,
  "metadata": {
    "rows_returned": 2,
    "execution_time_ms": 45
  }
}

Tool Discovery

Tool Registration & Discovery Flow

List all available tools:

## List tools
tools = await client.list_tools()

for tool in tools:
    print(f"Tool: {tool.name}")
    print(f"  Description: {tool.description}")
    print(f"  Input Schema: {json.dumps(tool.inputSchema, indent=2)}")

Response Format:

{
  "tools": [
    {
      "name": "chat",
      "description": "Chat with the AI agent",
      "inputSchema": {
        "type": "object",
        "properties": {...},
        "required": ["query"]
      }
    }
  ]
}

Tool Execution

Synchronous Execution

Standard request-response pattern:

result = await client.call_tool(
    name="chat",
    arguments={"query": "Hello!"}
)

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

Asynchronous Execution

For long-running tools:

## Start execution
task_id = await client.call_tool_async(
    name="execute_code",
    arguments={"code": long_running_code}
)

## Poll for completion
while True:
    status = await client.get_tool_status(task_id)
    if status.completed:
        print(status.result)
        break
    await asyncio.sleep(1)

Streaming Execution

For tools that support streaming:

async for chunk in client.call_tool_stream(
    name="chat",
    arguments={"query": "Write a long essay", "stream": True}
):
    print(chunk.delta, end="", flush=True)

Error Handling

Tools may return errors in the response:

{
  "content": [
    {
      "type": "text",
      "text": "Error: Invalid SQL syntax"
    }
  ],
  "isError": true,
  "metadata": {
    "error_code": "SYNTAX_ERROR",
    "error_details": "Unexpected token at position 10"
  }
}

Handle errors:

result = await client.call_tool("query_database", {
    "query": "INVALID SQL",
    "database": "users"
})

if result.isError:
    error_message = result.content[0].text
    error_code = result.metadata.get("error_code")
    print(f"Error ({error_code}): {error_message}")
else:
    print(result.content[0].text)

Custom Tools

Extend the server with custom tools:

Define Custom Tool

from langchain_core.tools import tool

@tool
def calculate_mortgage(
    principal: float,
    annual_rate: float,
    years: int
) -> str:
    """Calculate monthly mortgage payment.

    Args:
        principal: Loan amount in dollars
        annual_rate: Annual interest rate (e.g., 3.5 for 3.5%)
        years: Loan term in years
    """
    monthly_rate = annual_rate / 100 / 12
    num_payments = years * 12

    if monthly_rate == 0:
        monthly_payment = principal / num_payments
    else:
        monthly_payment = (
            principal * monthly_rate * (1 + monthly_rate) ** num_payments
        ) / ((1 + monthly_rate) ** num_payments - 1)

    return f"Monthly payment: ${monthly_payment:.2f}"

## Register tool
from mcp_server_langgraph.tools import register_tool

register_tool(calculate_mortgage)

Use Custom Tool

result = await client.call_tool("calculate_mortgage", {
    "principal": 300000,
    "annual_rate": 3.5,
    "years": 30
})

print(result.content[0].text)
## Output: Monthly payment: $1347.13

Tool Permissions

Tools respect authorization rules:

## Check if user can execute tool
allowed = await openfga_client.check_permission(
    user=f"user:{user_id}",
    relation="executor",
    object="tool:search_web"
)

if not allowed:
    raise PermissionError("User cannot execute search_web tool")

Grant permission:

## Grant user access to tool
await openfga_client.write_tuples([{
    "user": f"user:{user_id}",
    "relation": "executor",
    "object": "tool:search_web"
}])

Tool Composition

Chain multiple tools:

## 1. Search for information
search_result = await client.call_tool("search_web", {
    "query": "Python async best practices"
})

## 2. Summarize with chat
summary = await client.call_tool("chat", {
    "query": f"Summarize this:\n\n{search_result.content[0].text}"
})

print(summary.content[0].text)

Tool Monitoring

Track tool usage:

from prometheus_client import Counter, Histogram

tool_calls = Counter(
    'mcp_tool_calls_total',
    'Total tool calls',
    ['tool_name', 'status']
)

tool_duration = Histogram(
    'mcp_tool_duration_seconds',
    'Tool execution duration',
    ['tool_name']
)

## Track metrics
import time

@tool_duration.labels(tool_name="chat").time()
async def tracked_call_tool(name: str, arguments: dict):
    try:
        result = await client.call_tool(name, arguments)
        tool_calls.labels(tool_name=name, status="success").inc()
        return result
    except Exception as e:
        tool_calls.labels(tool_name=name, status="error").inc()
        raise

Best Practices

Validate Inputs

Always validate tool inputs against the schema:

from jsonschema import validate, ValidationError

try:
    validate(
        instance=arguments,
        schema=tool.inputSchema
    )
except ValidationError as e:
    raise ValueError(f"Invalid arguments: {e.message}")

Handle Timeouts

Set appropriate timeouts for tool execution:

import asyncio

try:
    result = await asyncio.wait_for(
        client.call_tool(name, arguments),
        timeout=30.0
    )
except asyncio.TimeoutError:
    print("Tool execution timed out")

Implement Retry Logic

Retry transient failures:

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def call_tool_with_retry(name: str, arguments: dict):
    return await client.call_tool(name, arguments)

Log Tool Usage

Log all tool executions for debugging and auditing:

logger.info(
    "Tool called",
    tool_name=name,
    user_id=user_id,
    arguments=mask_sensitive(arguments),
    timestamp=datetime.utcnow()
)

Troubleshooting

Tool not found

Error: Method not found: tool 'xyz' does not existSolutions:

# List available tools
tools = await client.list_tools()
tool_names = [t.name for t in tools]
print(f"Available tools: {tool_names}")

# Verify tool name (case-sensitive)
if "search_web" in tool_names:
    result = await client.call_tool("search_web", {...})

Permission denied

Error: User does not have permission to execute toolSolutions:

# Check permission
allowed = await openfga_client.check_permission(
    user=f"user:{user_id}",
    relation="executor",
    object=f"tool:{tool_name}"
)

# Grant if needed
if not allowed:
    await openfga_client.write_tuples([{
        "user": f"user:{user_id}",
        "relation": "executor",
        "object": f"tool:{tool_name}"
    }])

Invalid arguments

Error: Invalid params: required field 'query' missingSolutions:

# Check required fields
required = tool.inputSchema.get("required", [])
for field in required:
    if field not in arguments:
        raise ValueError(f"Missing required field: {field}")

# Validate against schema
from jsonschema import validate
validate(instance=arguments, schema=tool.inputSchema)

Next Steps

MCP Messages

Message protocol reference

MCP Resources

Resource types and access

MCP Endpoints

Available endpoints

Authorization

Tool permissions

MCP Tools Ready: Powerful, composable tools for AI agent capabilities!

API Documentation

Developer Tools

MCP Protocol

Overview

Available Tools

Agent Chat Tool

Conversation Search Tool

Conversation Get Tool

Execute Code Tool

Database Query Tool

Tool Discovery

Tool Registration & Discovery Flow

Tool Execution

Synchronous Execution

Asynchronous Execution

Streaming Execution

Error Handling

Custom Tools

Define Custom Tool

Use Custom Tool

Tool Permissions

Tool Composition

Tool Monitoring

Best Practices

Troubleshooting

Next Steps

MCP Messages

MCP Resources

MCP Endpoints

Authorization

​Overview

​Available Tools

​Agent Chat Tool

​Conversation Search Tool

​Conversation Get Tool

​Execute Code Tool

​Database Query Tool

​Tool Discovery

​Tool Registration & Discovery Flow

​Tool Execution

​Synchronous Execution

​Asynchronous Execution

​Streaming Execution

​Error Handling

​Custom Tools

​Define Custom Tool

​Use Custom Tool

​Tool Permissions

​Tool Composition

​Tool Monitoring

​Best Practices

​Troubleshooting

​Next Steps

MCP Messages

MCP Resources

MCP Endpoints

Authorization

Overview

Available Tools

Agent Chat Tool

Conversation Search Tool

Conversation Get Tool

Execute Code Tool

Database Query Tool

Tool Discovery

Tool Registration & Discovery Flow

Tool Execution

Synchronous Execution

Asynchronous Execution

Streaming Execution

Error Handling

Custom Tools

Define Custom Tool

Use Custom Tool

Tool Permissions

Tool Composition

Tool Monitoring

Best Practices

Troubleshooting

Next Steps