动态工具加载服务器

# Create a tool
mkdir -p tools
cat > tools/hello.py << 'EOF'
def greet(name: str) -> str:
    """Greet someone by name."""
    return f"Hello, {name}!"
EOF

# Run the server
docker run -d \
  -p 8000:8000 \
  -v $(pwd)/tools:/app/inline/tools \
  docker.io/helmforge/fastmcp-server:0.4.0

Sources (Inline, S3, Git)
        │
        ▼
  /app/workspace/
  ├── tools/*.py         → registered as MCP tools
  ├── resources/*.py     → registered as MCP resources
  ├── prompts/*.py       → registered as MCP prompts
  └── knowledge/*        → served as knowledge:// resources
        │
        ▼
  FastMCP Server (:8000/mcp)
  ├── /ui               → Web dashboard
  ├── /healthz          → Liveness probe
  ├── /readyz           → Readiness probe
  ├── /startupz         → Startup probe
  ├── /debug/info       → Server diagnostics
  ├── /metrics          → Prometheus metrics (opt-in)
  └── /api/*            → JSON API for UI

services:
  mcp-server:
    image: docker.io/helmforge/fastmcp-server:0.4.0
    ports:
      - "8000:8000"
    volumes:
      - ./tools:/app/inline/tools
      - ./resources:/app/inline/resources
      - ./prompts:/app/inline/prompts
      - ./knowledge:/app/inline/knowledge
    environment:
      MCP_SERVER_NAME: my-mcp-server
      MCP_AUTH_TYPE: bearer
      MCP_AUTH_TOKEN: ${MCP_AUTH_TOKEN}
    healthcheck:
      test: ["CMD", "curl", "-sf", "http://localhost:8000/healthz"]
      interval: 30s
      timeout: 5s
      retries: 3

services:
  mcp-server:
    image: docker.io/helmforge/fastmcp-server:0.4.0
    ports:
      - "8000:8000"
    environment:
      SOURCE_S3_ENABLED: "true"
      SOURCE_S3_ENDPOINT: http://minio:9000
      SOURCE_S3_BUCKET: mcp-tools
      SOURCE_S3_ACCESS_KEY: minioadmin
      SOURCE_S3_SECRET_KEY: minioadmin

  minio:
    image: docker.io/minio/minio:RELEASE.2025-04-03T14-56-28Z
    command: server /data --console-address ":9001"
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin

docker run -d \
  -p 8000:8000 \
  -e SOURCE_GIT_ENABLED=true \
  -e SOURCE_GIT_REPOSITORY=https://github.com/myorg/mcp-tools.git \
  -e SOURCE_GIT_BRANCH=main \
  docker.io/helmforge/fastmcp-server:0.4.0

def get_weather(city: str) -> str:
    """Get current weather for a city."""
    import httpx
    return httpx.get(f"https://wttr.in/{city}?format=3").text

def roll_dice(sides: int = 6) -> int:
    """Roll a die with the given number of sides."""
    import random
    return random.randint(1, sides)

__mcp_auto_register__ = False

def evidence_true(data, *keys):
    return True

TOOLS = ["deploy"]

def deploy(service: str, version: str) -> str:
    """Deploy a service."""
    return f"Deployed {service}@{version}"

def helper() -> str:
    return "not registered"

__tags__ = {"devops", "production"}        # Categorization tags
__timeout__ = 30.0                          # Execution timeout (seconds)
__annotations_mcp__ = {                     # MCP behavior hints
    "destructiveHint": True,
    "title": "Deploy Service"
}

def deploy(service: str, version: str) -> str:
    """Deploy a service to production."""
    return f"Deployed {service}@{version}"

async def fetch_url(url: str) -> str:
    """Fetch content from a URL."""
    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.text

from fastmcp.tools.tool import ToolResult

def analyze(data: str) -> ToolResult:
    """Analyze data with structured output."""
    return ToolResult(
        content="Analysis complete",
        structured_content={"word_count": len(data.split())},
        meta={"version": "1.0"}
    )

import json

RESOURCE_URI = "config://app"

def get_config() -> str:
    """Application configuration."""
    return json.dumps({"version": "1.0", "env": "production"}, indent=2)

import json

RESOURCE_URI = "users://{user_id}/profile"

def get_profile(user_id: str) -> str:
    """Get user profile by ID."""
    return json.dumps({"user_id": user_id, "name": f"User {user_id}"}, indent=2)

import json

RESOURCES = {
    "status://health": "get_health",
    "status://version": "get_version",
}

def get_health() -> str:
    return json.dumps({"status": "ok"}, indent=2)

def get_version() -> str:
    return "1.0.0"

def summarize(text: str) -> str:
    """Summarize the provided text."""
    return f"Please provide a concise summary of:\n\n{text}"

knowledge/
├── product-overview.md
├── api-reference.json
└── troubleshooting/
    ├── common-errors.md
    └── faq.md

{
  "timestamp": "2026-04-05T10:30:00+00:00",
  "level": "INFO",
  "logger": "fastmcp-server.builder",
  "message": "Registered tool: greet"
}

python /app/sync_only.py

{
  "mcpServers": {
    "my-mcp-server": {
      "type": "streamable-http",
      "url": "https://mcp.example.com/mcp"
    }
  }
}

{
  "mcpServers": {
    "my-mcp-server": {
      "type": "streamable-http",
      "url": "https://mcp.example.com/mcp",
      "headers": {
        "Authorization": "Bearer <your-token>"
      }
    }
  }
}

[mcp_servers.my-mcp-server]
enabled = true
url = "https://mcp.example.com/mcp"

[mcp_servers.my-mcp-server.http_headers]
Authorization = "Bearer <your-token>"

[mcp_servers.my-mcp-server]
enabled = true
url = "https://mcp.example.com/mcp"