pip install "mcp[cli]"

Successfully installed mcp-1.x.x ...

npm install @modelcontextprotocol/sdk

added 12 packages in 3s

go get github.com/mark3labs/mcp-go

go: added github.com/mark3labs/mcp-go v0.x.x

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("demo-server")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two integers and return the result."""
    return a + b

@mcp.tool()
def greet(name: str) -> str:
    """Return a friendly greeting for the given name."""
    return f"Hello, {name}!"

@mcp.resource("config://app")
def app_config() -> str:
    """Static application configuration."""
    return '{"version": "1.0", "feature_flags": {"beta": true}}'

@mcp.prompt()
def code_review(language: str, code: str) -> str:
    """Prompt template for reviewing code in the given language."""
    return f"Review the following {language} code for bugs and style:\n\n{code}"

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

python server.py

[INFO] FastMCP server "demo-server" running on stdio

+----------------+       JSON-RPC 2.0        +-----------------+
|   MCP client   | <-----------------------> |   MCP server    |
| (Claude Code,  |   stdio / SSE / HTTP      | (FastMCP, mcp-  |
|  Cursor, etc.) |                           |  go, custom)    |
+----------------+                           +-----------------+
       |                                              |
       | initialize → list_tools → call_tool ...      |
       v                                              v
+----------------+                           +-----------------+
|   Host LLM     |                           |  Your business  |
|   (Claude,     |                           |  logic / APIs   |
|   GPT, etc.)   |                           |                 |
+----------------+                           +-----------------+

from mcp.server.fastmcp import FastMCP
mcp = FastMCP("local")
if __name__ == "__main__":
    mcp.run(transport="stdio")

[INFO] FastMCP server "local" running on stdio

{
  "mcpServers": {
    "local": {
      "command": "python",
      "args": ["/abs/path/to/server.py"]
    }
  }
}

from mcp.server.fastmcp import FastMCP
mcp = FastMCP("remote-sse")
if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=8000)

[INFO] FastMCP server "remote-sse" listening at http://0.0.0.0:8000
[INFO] SSE endpoint:    http://0.0.0.0:8000/sse
[INFO] Message endpoint: http://0.0.0.0:8000/messages

from mcp.server.fastmcp import FastMCP
mcp = FastMCP("remote-http")
if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

[INFO] FastMCP server "remote-http" listening at http://0.0.0.0:8000/mcp

from typing import Annotated
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("github")

@mcp.tool()
def open_issue(
    repo: Annotated[str, "owner/name"],
    title: Annotated[str, "Issue title"],
    body:  Annotated[str, "Issue body in Markdown"] = "",
) -> str:
    """Open a new GitHub issue and return its URL."""
    ...
    return f"https://github.com/{repo}/issues/123"

@mcp.resource("docs://intro")
def docs_intro() -> str:
    """The intro page of the docs."""
    return "Welcome to the docs..."

@mcp.resource("file:///{path}")
def file_read(path: str) -> str:
    """Read an arbitrary file by absolute path."""
    with open(path) as f:
        return f.read()

@mcp.prompt()
def summarise(language: str) -> str:
    """Summarise a document in the given language."""
    return f"Summarise the attached document in {language}. Keep it under 200 words."

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather")

@mcp.tool()
async def get_weather(city: str) -> str:
    """Return current weather for a city."""
    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.get(f"https://wttr.in/{city}?format=3")
        return r.text

from mcp.server.fastmcp import Context, FastMCP

mcp = FastMCP("indexer")

@mcp.tool()
async def reindex(directory: str, ctx: Context) -> str:
    """Re-index all files in a directory, reporting progress."""
    files = list_files(directory)
    for i, f in enumerate(files):
        await ctx.info(f"indexing {f}")
        await ctx.report_progress(progress=i + 1, total=len(files))
        process(f)
    return f"Indexed {len(files)} files."

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({ name: "ts-demo", version: "1.0.0" });

server.registerTool(
  "multiply",
  {
    title: "Multiply",
    description: "Multiply two numbers.",
    inputSchema: { a: z.number(), b: z.number() },
  },
  async ({ a, b }) => ({
    content: [{ type: "text", text: String(a * b) }],
  }),
);

server.registerResource(
  "version",
  "app://version",
  { mimeType: "text/plain" },
  async () => ({ contents: [{ uri: "app://version", text: "1.0.0" }] }),
);

const transport = new StdioServerTransport();
await server.connect(transport);

package main

import (
    "context"
    "fmt"

    "github.com/mark3labs/mcp-go/mcp"
    "github.com/mark3labs/mcp-go/server"
)

func main() {
    s := server.NewMCPServer("go-demo", "1.0.0",
        server.WithToolCapabilities(true),
    )

    s.AddTool(mcp.NewTool("multiply",
        mcp.WithDescription("Multiply two integers."),
        mcp.WithNumber("a", mcp.Required(), mcp.Description("first operand")),
        mcp.WithNumber("b", mcp.Required(), mcp.Description("second operand")),
    ), func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
        a := req.GetFloat("a", 0)
        b := req.GetFloat("b", 0)
        return mcp.NewToolResultText(fmt.Sprintf("%g", a*b)), nil
    })

    server.ServeStdio(s)
}

go build -o go-mcp .
./go-mcp

(silent — stdio server waiting for client frames)

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

async def main():
    params = StdioServerParameters(command="python", args=["server.py"])
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print([t.name for t in tools.tools])
            result = await session.call_tool("add", {"a": 2, "b": 3})
            print(result.content[0].text)

asyncio.run(main())

['add', 'greet']
5

from mcp.server.fastmcp import FastMCP
from starlette.requests import Request

mcp = FastMCP("authed")

@mcp.tool()
async def whoami(ctx) -> str:
    """Return the authenticated user from the request token."""
    request: Request = ctx.request
    token = request.headers.get("authorization", "").removeprefix("Bearer ")
    return verify_jwt(token).sub

from mcp.client.streamable_http import streamablehttp_client

async with streamablehttp_client(
    url="https://example.com/mcp",
    headers={"Authorization": f"Bearer {token}"},
) as (read, write, _):
    ...

import os
import psycopg
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("postgres")
DSN = os.environ["DATABASE_URL"]

@mcp.tool()
def list_tables() -> list[str]:
    """List public tables in the connected Postgres database."""
    with psycopg.connect(DSN) as cn, cn.cursor() as cur:
        cur.execute("select table_name from information_schema.tables where table_schema = 'public'")
        return [r[0] for r in cur.fetchall()]

@mcp.tool()
def query(sql: str) -> list[dict]:
    """Run a read-only SQL query and return rows as dicts."""
    if not sql.lstrip().lower().startswith("select"):
        raise ValueError("Only SELECT queries are allowed.")
    with psycopg.connect(DSN) as cn, cn.cursor(row_factory=psycopg.rows.dict_row) as cur:
        cur.execute(sql)
        return cur.fetchmany(100)

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

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("jira")
BASE = os.environ["JIRA_BASE_URL"]
TOKEN = os.environ["JIRA_TOKEN"]

client = httpx.AsyncClient(base_url=BASE, headers={"Authorization": f"Bearer {TOKEN}"})

@mcp.tool()
async def search_issues(jql: str) -> list[dict]:
    """Search Jira issues using JQL."""
    r = await client.get("/rest/api/3/search", params={"jql": jql, "maxResults": 20})
    return r.json().get("issues", [])

if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8001)

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

async def gather_tools():
    servers = {
        "fs":     StdioServerParameters(command="python", args=["./fs_server.py"]),
        "github": StdioServerParameters(command="python", args=["./gh_server.py"]),
        "db":     StdioServerParameters(command="python", args=["./db_server.py"]),
    }
    sessions = {}
    all_tools = []
    for name, params in servers.items():
        read, write = await stdio_client(params).__aenter__()
        session = ClientSession(read, write)
        await session.__aenter__()
        await session.initialize()
        tools = (await session.list_tools()).tools
        for t in tools:
            t.name = f"{name}.{t.name}"
        all_tools.extend(tools)
        sessions[name] = session
    return sessions, all_tools

from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, ToolError

mcp = FastMCP("billing")

@mcp.tool()
def charge(customer_id: str, amount_cents: int) -> list[TextContent]:
    """Charge a customer's saved payment method."""
    if amount_cents <= 0:
        raise ToolError("amount_cents must be positive")
    txn_id = stripe_charge(customer_id, amount_cents)
    return [TextContent(type="text", text=f"Charged. Transaction {txn_id}.")]

pip install "mcp[cli]"
mcp dev server.py

Inspector running at http://localhost:5173
Server stdio attached.