MCP &
Protocols

pip install "mcp[cli]"

from mcp.server.fastmcp import FastMCP

# Create the server — give it a name clients will see
mcp = FastMCP("course-tools")

# ── In-memory note store ─────────────────────────────────────────
_notes: list[str] = []


# ── Tool 1: Calculator ───────────────────────────────────────────
@mcp.tool()
def calculator(expression: str) -> str:
    """Evaluate a Python arithmetic expression and return the result.

    Args:
        expression: A safe arithmetic expression, e.g. '(12 + 4) * 3'
    """
    try:
        result = eval(expression, {"__builtins__": {}})
        return str(result)
    except Exception as e:
        return f"Error: {e}"


# ── Tool 2: Add a note ───────────────────────────────────────────
@mcp.tool()
def add_note(note: str) -> str:
    """Save a note to the in-memory note store.

    Args:
        note: The text content of the note to save.
    """
    _notes.append(note)
    return f"Note saved. Total notes: {len(_notes)}"


# ── Tool 3: List notes ───────────────────────────────────────────
@mcp.tool()
def list_notes() -> str:
    """Return all saved notes as a numbered list."""
    if not _notes:
        return "No notes saved yet."
    return "\n".join(f"{i+1}. {n}" for i, n in enumerate(_notes))

# ── Resource: notes list ─────────────────────────────────────────
@mcp.resource("notes://all")
def get_all_notes() -> str:
    """All saved notes, readable as a plain-text resource.

    URI: notes://all
    MIME type: text/plain
    """
    if not _notes:
        return "(no notes saved)"
    return "\n".join(f"{i+1}. {n}" for i, n in enumerate(_notes))


# ── Start the server (stdio transport) ───────────────────────────
if __name__ == "__main__":
    mcp.run()  # defaults to stdio transport

mcp dev mcp_server.py

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


async def main():
    # Launch the server as a subprocess (stdio transport)
    server_params = StdioServerParameters(
        command="python",
        args=["mcp_server.py"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Handshake — negotiate capabilities
            await session.initialize()

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

            # Call calculator tool
            result = await session.call_tool("calculator", {"expression": "(365 * 24 * 60 * 60)"})
            print(f"\ncalculator('365 * 24 * 60 * 60') = {result.content[0].text}")

            # Save a couple of notes
            await session.call_tool("add_note", {"note": "MCP uses JSON-RPC 2.0 over stdio or HTTP/SSE."})
            await session.call_tool("add_note", {"note": "Tools are model-controlled; Resources are application-controlled."})

            # Read the resource
            resource = await session.read_resource("notes://all")
            print(f"\nnotes://all resource:\n{resource.contents[0].text}")


asyncio.run(main())

python mcp_client.py

Available tools:
  - calculator: Evaluate a Python arithmetic expression and return t
  - add_note: Save a note to the in-memory note store.
  - list_notes: Return all saved notes as a numbered list.

calculator('365 * 24 * 60 * 60') = 31536000

notes://all resource:
1. MCP uses JSON-RPC 2.0 over stdio or HTTP/SSE.
2. Tools are model-controlled; Resources are application-controlled.

import asyncio, os, json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import anthropic

anthropic_client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])


async def run_mcp_agent(task: str):
    server_params = StdioServerParameters(command="python", args=["mcp_server.py"])

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 1. Discover tools from MCP server
            mcp_tools = await session.list_tools()

            # 2. Convert to Anthropic tool schema format
            anthropic_tools = [
                {
                    "name": t.name,
                    "description": t.description,
                    "input_schema": t.inputSchema,
                }
                for t in mcp_tools.tools
            ]

            # 3. Run a simple agent loop
            messages = [{"role": "user", "content": task}]
            for _ in range(5):
                response = anthropic_client.messages.create(
                    model="claude-haiku-4-5-20251001",
                    max_tokens=512,
                    tools=anthropic_tools,
                    messages=messages,
                )
                messages.append({"role": "assistant", "content": response.content})

                if response.stop_reason == "end_turn":
                    print(f"\nANSWER: {response.content[0].text}")
                    break

                if response.stop_reason == "tool_use":
                    tool_results = []
                    for block in response.content:
                        if block.type == "tool_use":
                            # 4. Execute via MCP session instead of local executor
                            result = await session.call_tool(block.name, block.input)
                            tool_results.append({
                                "type": "tool_result",
                                "tool_use_id": block.id,
                                "content": result.content[0].text,
                            })
                    messages.append({"role": "user", "content": tool_results})


asyncio.run(run_mcp_agent(
    "Calculate 2 to the power of 16, then save a note with the result."
))