First Tool Invocation

Call your first MCP tool and understand the request/response flow.

Overview

MCP tools follow a simple request/response pattern:

Listing Available Tools

Before calling tools, see what's available:

from gateflow_mcp import MCPClient

client = MCPClient(
    agent_id="agent_abc123",
    api_key="gf-agent-xyz789..."
)

# List all tools
tools = client.list_tools()

for tool in tools:
    print(f"Tool: {tool.name}")
    print(f"  Description: {tool.description}")
    print(f"  Parameters: {tool.input_schema}")
    print()

Output:

Tool: llm/chat
  Description: Send messages to an LLM and get a response
  Parameters: {messages: array, model: string, temperature: number}

Tool: retrieval/search
  Description: Search the knowledge base for relevant documents
  Parameters: {query: string, limit: number, filters: object}

Tool: self_inspect/whoami
  Description: Get information about the current agent
  Parameters: {}

Calling a Tool

Basic Tool Call

# Call the search tool
result = client.call_tool(
    name="retrieval/search",
    arguments={
        "query": "password reset instructions",
        "limit": 5
    }
)

print(f"Found {len(result['results'])} results:")
for r in result["results"]:
    print(f"  - {r['title']} (score: {r['score']:.2f})")

Output:

Found 3 results:
  - Password Reset Guide (score: 0.95)
  - Account Recovery FAQ (score: 0.82)
  - Security Best Practices (score: 0.71)

LLM Chat Tool

result = client.call_tool(
    name="llm/chat",
    arguments={
        "messages": [
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": "What is the capital of France?"}
        ],
        "model": "gpt-5-mini",
        "temperature": 0.7
    }
)

print(result["content"])
# Output: The capital of France is Paris.

Voice Transcription Tool

import base64

# Read audio file
with open("audio.mp3", "rb") as f:
    audio_b64 = base64.b64encode(f.read()).decode()

result = client.call_tool(
    name="voice/transcribe",
    arguments={
        "audio": audio_b64,
        "model": "whisper-1",
        "language": "en"
    }
)

print(f"Transcription: {result['text']}")

Self-Inspect Tool

# Check agent identity and permissions
result = client.call_tool(name="self_inspect/whoami", arguments={})

print(f"Agent ID: {result['agent_id']}")
print(f"Name: {result['name']}")
print(f"Permissions: {result['permissions']}")
print(f"Session cost: ${result['session']['cost']:.4f}")

Response Format

All tool responses include:

{
  "result": {
    // Tool-specific response data
  },
  "metadata": {
    "tool": "retrieval/search",
    "latency_ms": 234,
    "cost": 0.0012,
    "audit_id": "audit_xyz789"
  }
}

Error Handling

Permission Denied

try:
    result = client.call_tool(
        name="voice/synthesize",
        arguments={"text": "Hello"}
    )
except MCPError as e:
    if e.code == "permission_denied":
        print(f"Agent lacks permission for: {e.tool}")
        print(f"Required permission: {e.required_permission}")

Invalid Arguments

try:
    result = client.call_tool(
        name="llm/chat",
        arguments={"messages": "not an array"}  # Wrong type
    )
except MCPError as e:
    if e.code == "invalid_arguments":
        print(f"Validation error: {e.message}")
        print(f"Schema: {e.schema}")

Rate Limit

try:
    result = client.call_tool(name="llm/chat", arguments={...})
except MCPError as e:
    if e.code == "rate_limit_exceeded":
        print(f"Rate limited. Retry after: {e.retry_after}s")

Cost Limit

try:
    result = client.call_tool(name="voice/transcribe", arguments={...})
except MCPError as e:
    if e.code == "cost_limit_exceeded":
        print(f"Cost limit reached: ${e.current:.2f} / ${e.limit:.2f}")

Using with LangChain

from langchain_mcp import MCPToolkit

toolkit = MCPToolkit(
    server_url="https://mcp.gateflow.ai/agent_abc123",
    api_key="gf-agent-xyz789..."
)

tools = toolkit.get_tools()

# Find specific tool
search_tool = next(t for t in tools if t.name == "retrieval_search")

# Invoke directly
result = search_tool.invoke({"query": "password reset"})
print(result)

Async Tool Calls

import asyncio
from gateflow_mcp import AsyncMCPClient

async def main():
    async with AsyncMCPClient(
        agent_id="agent_abc123",
        api_key="gf-agent-xyz789..."
    ) as client:
        # Parallel tool calls
        results = await asyncio.gather(
            client.call_tool("retrieval/search", {"query": "pricing"}),
            client.call_tool("retrieval/search", {"query": "features"}),
            client.call_tool("self_inspect/get_my_usage", {})
        )

        for result in results:
            print(result)

asyncio.run(main())

Tool Categories

Category	Tools	Description
voice/*	transcribe, synthesize, pipeline	Audio processing
document/*	ocr, process, status	Document handling
retrieval/*	search, rerank, search_and_rerank	Knowledge base
llm/*	chat, embed, list_models	AI model access
self_inspect/*	whoami, get_my_usage	Agent introspection

Best Practices

1. Check permissions first - Use self_inspect/whoami to see available tools
2. Handle errors gracefully - Wrap calls in try/catch
3. Monitor costs - Check usage regularly with self_inspect/get_my_usage
4. Use appropriate models - Match model to task complexity
5. Batch when possible - Reduce API calls with batch operations

Next Steps

• Tool Catalog - Explore all available tools
• Permissions - Understand the permission model
• Cost Transparency - Monitor agent costs