Building Agent Tools with FastMCP

# !pip install fastmcp pydantic-ai networkx quantecon mesa

from fastmcp import FastMCP

mcp = FastMCP("Calculator")

@mcp.tool()
def add(a: float, b: float) -> float:
    """Add two numbers together."""
    return a + b

@mcp.tool()
def multiply(a: float, b: float) -> float:
    """Multiply two numbers together."""
    return a * b

print("✓ Calculator MCP server created with 2 tools")

%%file calculator_server.py

from fastmcp import FastMCP

mcp = FastMCP("Calculator")

@mcp.tool()
def add(a: float, b: float) -> float:
    """Add two numbers together."""
    return a + b

if __name__ == "__main__":
    mcp.run()  # stdio by default

from fastmcp import Client

async with Client("calculator_server.py") as client:
    tools = await client.list_tools()
    print("Available tools:", [t.name for t in tools])

    result = await client.call_tool("add", {"a": 5, "b": 3})
    print("5 + 3 =", result)

%%file network_analysis_server.py

from fastmcp import FastMCP
import networkx as nx
from typing import Dict, List, Tuple, Any

# Global cache for persistent state across tool calls
# MCP Context is request-scoped, so we need external storage
cache: Dict[str, Any] = {}

network_mcp = FastMCP("NetworkAnalysis")

@network_mcp.tool()
def create_network(
    graph_id: str,
    edges: List[Tuple[int, int]]
) -> Dict[str, Any]:
    """
    Create a network from an edge list and store it.

    Args:
        graph_id: Unique identifier for this graph (e.g., 'social_network', 'graph1')
        edges: List of edges as [source, target] pairs. Example: [[1,2], [2,3], [1,3]]

    Returns:
        Dictionary with graph statistics (num_nodes, num_edges, density)
    """
    G = nx.Graph()
    G.add_edges_from(edges)
    cache[f"graph:{graph_id}"] = G
    return {
        "graph_id": graph_id,
        "num_nodes": G.number_of_nodes(),
        "num_edges": G.number_of_edges(),
        "density": round(nx.density(G), 4)
    }

@network_mcp.tool()
def calculate_degree_centrality(
    graph_id: str,
    node: int
) -> Dict[str, Any]:
    """
    Calculate degree centrality for a node.

    Degree centrality measures how many connections a node has.
    Higher values indicate more central/connected nodes.

    Args:
        graph_id: ID of the graph to analyze
        node: The node ID to calculate centrality for

    Returns:
        Dictionary with degree and normalized centrality value
    """
    G = cache.get(f"graph:{graph_id}")
    if G is None:
        return {"error": f"Graph '{graph_id}' not found. Create it first."}
    if node not in G:
        return {"error": f"Node {node} not in graph '{graph_id}'"}

    degree = G.degree(node)
    max_possible = G.number_of_nodes() - 1
    normalized = degree / max_possible if max_possible > 0 else 0
    return {"node": node, "degree": degree, "normalized_centrality": round(normalized, 4)}

@network_mcp.tool()
def calculate_betweenness(
    graph_id: str,
    node: int
) -> Dict[str, Any]:
    """
    Calculate betweenness centrality for a node.

    Betweenness measures how often a node lies on shortest paths between other nodes.
    High betweenness nodes are 'bridges' connecting different parts of the network.

    Args:
        graph_id: ID of the graph to analyze
        node: The node ID to calculate betweenness for

    Returns:
        Dictionary with betweenness centrality value
    """
    G = cache.get(f"graph:{graph_id}")
    if G is None:
        return {"error": f"Graph '{graph_id}' not found"}
    if node not in G:
        return {"error": f"Node {node} not in graph '{graph_id}'"}

    betweenness = nx.betweenness_centrality(G)
    return {"node": node, "betweenness_centrality": round(betweenness[node], 4)}

@network_mcp.tool()
def find_shortest_path(
    graph_id: str,
    source: int,
    target: int
) -> Dict[str, Any]:
    """
    Find shortest path between two nodes.

    Args:
        graph_id: ID of the graph to search
        source: Starting node ID
        target: Destination node ID

    Returns:
        Dictionary with path and length, or error if no path exists
    """
    G = cache.get(f"graph:{graph_id}")
    if G is None:
        return {"error": f"Graph '{graph_id}' not found"}

    try:
        path = nx.shortest_path(G, source, target)
        return {"found": True, "path": path, "length": len(path) - 1}
    except nx.NetworkXNoPath:
        return {"found": False, "message": f"No path exists between {source} and {target}"}
    except nx.NodeNotFound:
        return {"found": False, "message": f"One or both nodes not in graph"}

if __name__ == "__main__":
    network_mcp.run()

from fastmcp import Client

async with Client("network_analysis_server.py") as client:
    tools = await client.list_tools()
    print("Available tools:", [t.name for t in tools])
    
    # Create a social network
    result = await client.call_tool(
        "create_network",
        {"graph_id": "social", "edges": [[1,2], [1,3], [2,3], [3,4], [4,5], [5,6], [6,4], [3,7], [7,8], [8,9], [9,7]]}
    )
    print(f"\nNetwork created: {result.data}")
    
    # Analyze centrality
    result = await client.call_tool("calculate_degree_centrality", {"graph_id": "social", "node": 3})
    print(f"Node 3 centrality: {result.data}")
    
    result = await client.call_tool("calculate_betweenness", {"graph_id": "social", "node": 3})
    print(f"Node 3 betweenness: {result.data}")
    
    # Find path
    result = await client.call_tool("find_shortest_path", {"graph_id": "social", "source": 1, "target": 9})
    print(f"Path 1→9: {result.data}")

%%file game_server.py

from fastmcp import FastMCP
import quantecon.game_theory as gt
import numpy as np
from typing import Dict, List, Any

game_mcp = FastMCP("GameTheory")
game_cache: Dict[str, Any] = {}

@game_mcp.tool()
def create_game(
    game_id: str,
    payoff_matrix_p1: List[List[float]],
    payoff_matrix_p2: List[List[float]]
) -> Dict[str, Any]:
    """
    Create a two-player normal-form game.

    Args:
        game_id: Unique identifier for this game
        payoff_matrix_p1: Payoff matrix for Player 1 (rows = P1 strategies, cols = P2 strategies)
        payoff_matrix_p2: Payoff matrix for Player 2 (rows = P1 strategies, cols = P2 strategies)

    Returns:
        Game statistics and confirmation
    """
    p1_payoffs = np.array(payoff_matrix_p1)
    p2_payoffs = np.array(payoff_matrix_p2)
    game = gt.NormalFormGame([p1_payoffs, p2_payoffs])
    game_cache[game_id] = game
    return {
        "game_id": game_id,
        "num_players": 2,
        "p1_strategies": p1_payoffs.shape[0],
        "p2_strategies": p1_payoffs.shape[1],
        "message": f"Game '{game_id}' created successfully"
    }

@game_mcp.tool()
def find_pure_nash_equilibria(game_id: str) -> Dict[str, Any]:
    """
    Find all pure strategy Nash equilibria in the game.

    A Nash equilibrium is a strategy profile where no player can improve
    by unilaterally changing their strategy.

    Args:
        game_id: ID of the game to analyze

    Returns:
        List of Nash equilibria (strategy profiles) and their payoffs
    """
    game = game_cache.get(game_id)
    if game is None:
        return {"error": f"Game '{game_id}' not found"}

    equilibria = game.pure_nash_brute()
    results = []
    for eq in equilibria:
        payoffs = [game.players[i].payoff_array[eq] for i in range(len(game.players))]
        results.append({"strategies": eq, "payoffs": [float(p) for p in payoffs]})

    return {
        "game_id": game_id,
        "num_equilibria": len(results),
        "equilibria": results
    }

print("✓ Game Theory MCP server created")

%%file abm_server.py

from fastmcp import FastMCP
from mesa import Agent, Model
from mesa.time import RandomActivation
from mesa.space import SingleGrid
from mesa.datacollection import DataCollector
from typing import Dict, Any
import random

abm_mcp = FastMCP("AgentBasedModels")
abm_cache: Dict[str, Any] = {}

class SchellingAgent(Agent):
    def __init__(self, unique_id, model, agent_type):
        super().__init__(unique_id, model)
        self.type = agent_type

    def step(self):
        neighbors = self.model.grid.get_neighbors(self.pos, moore=True, include_center=False)
        similar = sum(1 for n in neighbors if n.type == self.type)
        total = len(neighbors)
        if total > 0 and (similar / total) < self.model.homophily:
            self.model.grid.move_to_empty(self)

class SchellingModel(Model):
    def __init__(self, width=20, height=20, density=0.8, minority_pc=0.2, homophily=3):
        super().__init__()
        self.homophily = homophily / 8
        self.schedule = RandomActivation(self)
        self.grid = SingleGrid(width, height, torus=True)
        n_agents = int(width * height * density)
        for i in range(n_agents):
            agent_type = 1 if random.random() < minority_pc else 0
            agent = SchellingAgent(i, self, agent_type)
            self.schedule.add(agent)
            self.grid.position_agent(agent, (random.randrange(width), random.randrange(height)))
        self.datacollector = DataCollector(model_reporters={"segregation": lambda m: self.measure_segregation(m)})

    @staticmethod
    def measure_segregation(model):
        similar_neighbors = total_neighbors = 0
        for agent in model.schedule.agents:
            neighbors = model.grid.get_neighbors(agent.pos, moore=True, include_center=False)
            if neighbors:
                similar_neighbors += sum(1 for n in neighbors if n.type == agent.type)
                total_neighbors += len(neighbors)
        return similar_neighbors / total_neighbors if total_neighbors > 0 else 0

    def step(self):
        self.datacollector.collect(self)
        self.schedule.step()

@abm_mcp.tool()
def create_schelling_model(
    model_id: str,
    width: int = 20,
    height: int = 20,
    density: float = 0.8,
    minority_percent: float = 0.2,
    homophily: int = 3
) -> Dict[str, Any]:
    """
    Create a Schelling segregation model.

    The Schelling model demonstrates how mild preferences for similar neighbors
    can lead to high levels of segregation.

    Args:
        model_id: Unique identifier for this model
        width: Grid width (default 20)
        height: Grid height (default 20)
        density: Fraction of cells occupied (0-1, default 0.8)
        minority_percent: Fraction of agents that are minority type (0-1, default 0.2)
        homophily: Number of similar neighbors desired (out of 8, default 3)

    Returns:
        Model configuration and initial state
    """
    model = SchellingModel(width, height, density, minority_percent, homophily)
    abm_cache[model_id] = model
    return {
        "model_id": model_id,
        "width": width,
        "height": height,
        "num_agents": len(model.schedule.agents),
        "initial_segregation": round(model.measure_segregation(model), 3)
    }

@abm_mcp.tool()
def step_model(
    model_id: str,
    num_steps: int = 1
) -> Dict[str, Any]:
    """
    Run the model for a specified number of steps.

    Args:
        model_id: ID of the model to step
        num_steps: Number of steps to run (default 1)

    Returns:
        Segregation metrics after stepping
    """
    model = abm_cache.get(model_id)
    if model is None:
        return {"error": f"Model '{model_id}' not found"}

    for _ in range(num_steps):
        model.step()

    df = model.datacollector.get_model_vars_dataframe()
    return {
        "model_id": model_id,
        "steps_completed": num_steps,
        "total_steps": len(df),
        "current_segregation": round(df['segregation'].iloc[-1], 3),
        "initial_segregation": round(df['segregation'].iloc[0], 3)
    }

print("✓ ABM MCP server created")

from fastmcp import FastMCP, Context

network_resources_mcp = FastMCP("NetworkResources")

@network_resources_mcp.resource("network://{graph_id}/summary")
def get_network_summary(ctx: Context, graph_id: str) -> str:
    """Get a text summary of network properties."""
    if not hasattr(ctx, 'graphs') or graph_id not in ctx.graphs:
        return f"Error: Graph '{graph_id}' not found"
    G = ctx.graphs[graph_id]
    return f"Network {graph_id}: {G.number_of_nodes()} nodes, {G.number_of_edges()} edges, density {nx.density(G):.4f}"

print("✓ Network resources server created")

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
from dotenv import load_dotenv

load_dotenv()

from network_analysis_server import network_mcp
toolset = MCPToolset(network_mcp)
agent = Agent('anthropic:claude-haiku-4-5', toolsets=[toolset])

async def main():
    prompt = """
    Create a social network with friendships: 1↔2, 1↔3, 1↔4, 2↔3, 3↔4.
    Person 5 is isolated. Analyze the structure.
    """
    result = await agent.run(prompt)
    print(result.output)

await main()

# Example with file-based server
toolset = MCPToolset('network_analysis_server.py')
agent = Agent('anthropic:claude-haiku-4-5', toolsets=[toolset], system_prompt="You are a helpful network analysis assistant.")

async with agent:
    result = await agent.run('Create a path network (1-2-3-4-5) and find the shortest path from 1 to 5')
    print("Agent Response:", result.output)

from fastmcp import FastMCP

network_prompts_mcp = FastMCP("NetworkPrompts")

@network_prompts_mcp.prompt()
def analyze_network_structure(graph_id: str) -> str:
    """Generate a comprehensive network analysis prompt."""
    return f"""
Please analyze the network '{graph_id}':
1. Basic Statistics (nodes, edges, density)
2. Centrality Analysis (top nodes by degree and betweenness)
3. Structural Properties (connectivity, clustering)
4. Interpretation (information flow implications)
"""

print("✓ Network prompts server created")

from fastmcp import FastMCP
from typing import List

stats_mcp = FastMCP("Statistics")

@stats_mcp.tool()
def calculate_mean(data: List[float]) -> float:
    """Calculate the arithmetic mean of a list of numbers."""
    # TODO: implement
    pass

# TODO: Add other tools

from typing import Dict

@network_mcp.tool()
def calculate_clustering_coefficient(ctx: Context, graph_id: str) -> Dict[str, any]:
    """Calculate the average clustering coefficient."""
    # TODO: implement
    pass