Agentic AI – MCP Custom Server Implementation


Agentic AI – MCP Custom Server Implementation

Table Of Contents:

  1. MCP Server Implementation?
  2. Math MCP Server Implementation.
  3. Weather MCP Server Implementation.
  4. MultiServer MCP Client + LangChain Agent.
  5. Run The Workflow.

(1) MCP Server Implementation

(2) Math MCP Server Implementation

# math_server.py
from fastmcp import FastMCP 

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b:int)->int:
    """Add Two Numbers"""
    return a * b 
    
@mcp.tool()
def multiply(a:int, b:int)-> int:
    """Multiply Two Numbers"""
    return a * b 
    
if __name__ ==== "__main__":
   mcp.run(transport="stdio")

(3) Weather MCP Server Implementation

# weather_server.py
from fastamcp import FastMCP 

mcp = FastMCP("Weather")

@mcp.tool()
async def get_weather(location: str) -> str:
     """Get Weather For Location"""
     return f"Its always Sunny in {location}"
     
if __name__ == "__main__":
   mcp.run(transport = "streamable-http")

(4) MultiServer MCP Client + LangChain Agent

# client.py
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent

async def main():
    # Connect to both servers
    client = MultiServerMCPClient(
        {
            "math": {
                "transport": "stdio",
                "command": "python",
                "args": ["/absolute/path/to/math_server.py"],
            },
            "weather": {
                "transport": "http",
                "url": "http://localhost:8000/mcp",  # Weather server endpoint
            }
        }
    )

    # Discover tools from servers
    tools = await client.get_tools()

    # Create agent with Claude Sonnet + MCP tools
    agent = create_agent(
        "claude-sonnet-4-6",
        tools
    )

    # Example 1: Math query
    math_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "what's (3 + 5) x 12?"}]}
    )

    # Example 2: Weather query
    weather_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "what is the weather in nyc?"}]}
    )

    print("Math Response:", math_response)
    print("Weather Response:", weather_response)

    # Close connections
    await client.close()

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

(5) Run The Workflow.

Leave a Reply

Your email address will not be published. Required fields are marked *