	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > Model Context Protocol (MCP) - Connecting AI Agents to External Tools

Model Context Protocol (MCP) - Connecting AI Agents to External Tools

Author: Venkata Sudhakar

Model Context Protocol (MCP) is an open standard introduced by Anthropic in 2024 that defines a universal way for AI models to connect to external tools, data sources, and services. Before MCP, every AI application had to implement its own custom integration layer for each tool - one custom adapter for GitHub, another for Slack, another for a database, and so on. MCP standardises this with a client-server architecture: any application that implements the MCP client specification can connect to any MCP server, regardless of which AI model or framework is being used.

An MCP server exposes three types of capabilities: Tools (functions the AI can call, like "search_github" or "run_sql_query"), Resources (data the AI can read, like file contents or database records), and Prompts (reusable prompt templates with parameters). MCP servers communicate over standard transports - either stdio (local process) or HTTP with SSE (remote server). The AI model host (like Claude Desktop, Cursor, or a custom application) acts as the MCP client, discovers available tools from connected servers, and decides when to call them based on user requests.

The below example shows how to build a custom MCP server in Python using the official MCP SDK that exposes two tools: one for querying database migration status and one for listing Kafka topic offsets.

# pip install mcp
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import json

# Create the MCP server instance
server = Server("migration-ops-server")

# Register Tool 1: Get migration job status
@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_migration_status",
            description="Get the current status of a database migration job. "
                        "Returns progress, row counts, and estimated completion time.",
            inputSchema={
                "type": "object",
                "properties": {
                    "job_id": {
                        "type": "string",
                        "description": "Migration job ID, e.g. MIG-1042"
                    }
                },
                "required": ["job_id"]
            }
        ),
        Tool(
            name="list_kafka_consumer_lag",
            description="List consumer group lag for all Kafka topics. "
                        "Returns topic, partition, current offset, end offset and lag.",
            inputSchema={
                "type": "object",
                "properties": {
                    "group_id": {
                        "type": "string",
                        "description": "Kafka consumer group ID to check"
                    }
                },
                "required": ["group_id"]
            }
        )
    ]

# Handle tool calls from the AI
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list:
    if name == "get_migration_status":
        job_id = arguments["job_id"]
        # In production: query your migration platform API
        mock_jobs = {
            "MIG-1042": {"status": "IN_PROGRESS", "rows_done": 2450000,
                         "total_rows": 5000000, "eta_min": 45},
            "MIG-1041": {"status": "COMPLETED", "rows_done": 1200000,
                         "total_rows": 1200000, "eta_min": 0},
            "MIG-1043": {"status": "FAILED", "error": "Schema mismatch on table orders"}
        }
        result = mock_jobs.get(job_id, {"status": "NOT_FOUND"})
        return [TextContent(type="text", text=json.dumps(result, indent=2))]

elif name == "list_kafka_consumer_lag":
        group_id = arguments["group_id"]
        # In production: use kafka-consumer-groups.sh or AdminClient
        mock_lag = [
            {"topic": "order-events", "partition": 0, "current": 45231, "end": 45234, "lag": 3},
            {"topic": "order-events", "partition": 1, "current": 38900, "end": 38900, "lag": 0},
            {"topic": "payment-events", "partition": 0, "current": 12100, "end": 12105, "lag": 5}
        ]
        return [TextContent(type="text", text=json.dumps({
            "group_id": group_id, "topics": mock_lag
        }, indent=2))]

return [TextContent(type="text", text=f"Unknown tool: {name}")]

# Run the server over stdio transport
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream,
                        server.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

# claude_desktop_config.json
{
  "mcpServers": {
    "migration-ops": {
      "command": "python",
      "args": ["/path/to/migration_mcp_server.py"]
    }
  }
}

# After restarting Claude Desktop, the AI can now:
User: "What is the status of migration job MIG-1042?"
Claude: [calls get_migration_status(job_id="MIG-1042")]
Claude: "Migration job MIG-1042 is currently IN PROGRESS. So far 2,450,000 out
of 5,000,000 rows have been migrated (49%), with approximately 45 minutes remaining."

User: "Is the Kafka consumer group cdc-sync-group keeping up?"
Claude: [calls list_kafka_consumer_lag(group_id="cdc-sync-group")]
Claude: "The cdc-sync-group is mostly keeping up. The order-events topic partition 0
has a lag of 3 messages and payment-events has a lag of 5 messages, both
within acceptable real-time bounds. Partition 1 of order-events is fully caught up."

# Testing your MCP server locally with the MCP inspector
# Run this command to launch the inspector UI:
# npx @modelcontextprotocol/inspector python migration_mcp_server.py

# Or test programmatically with the MCP client:
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio, json

async def test_mcp_server():
    server_params = StdioServerParameters(
        command="python",
        args=["migration_mcp_server.py"]
    )
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

# List available tools
            tools = await session.list_tools()
            print("Available tools:")
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description[:60]}...")

# Call a tool directly
            result = await session.call_tool(
                "get_migration_status",
                {"job_id": "MIG-1042"}
            )
            print("\nTool result:")
            print(result.content[0].text)

asyncio.run(test_mcp_server())

It gives the following output,

Available tools:
  - get_migration_status: Get the current status of a database migration job...
  - list_kafka_consumer_lag: List consumer group lag for all Kafka topics...

Tool result:
{
  "status": "IN_PROGRESS",
  "rows_done": 2450000,
  "total_rows": 5000000,
  "eta_min": 45
}

Why MCP matters for enterprise AI adoption:

MCP solves the "N x M integration problem." Without MCP, connecting N AI applications to M tools requires N x M custom integrations. With MCP, each tool is written once as an MCP server (M servers total) and each AI application is written once as an MCP client (N clients). Any client connects to any server automatically. For enterprise data migration and operations teams, this means you can build one MCP server for your migration platform, one for your Kafka cluster, one for your Kubernetes monitoring - and immediately make all of these available to any MCP-compatible AI tool your team uses, whether that is Claude Desktop, Cursor, or a custom LangChain application.

Send your comments, suggestions or queries regarding this site to [email protected].