GitHub - standup-coder/mcp4coder: mcp4coder

                    ┌───────────────┐
                    │  Web UI /ui   │
                    │  Swagger /docs│
                    └───────┬───────┘
                            │
                    ┌───────▼───────┐
                    │   FastAPI     │
                    │   REST API    │
                    │  Auth + CORS  │
                    └───────┬───────┘
                            │
              ┌─────────────┼─────────────┐
              │             │             │
     ┌────────▼──────┐ ┌───▼───┐ ┌──────▼──────┐
     │ ServerFactory │ │Service│ │  Celery     │
     │ 26 Servers    │ │ Layer │ │  Workers    │
     └────────┬──────┘ └───────┘ └──────┬──────┘
              │                         │
     ┌────────▼──────┐          ┌──────▼──────┐
     │ BaseMCPServer │          │    Redis    │
     │ register_tools│          │  Broker +   │
     │ execute_tool  │          │  Result     │
     └───────────────┘          └─────────────┘

git clone https://github.com/standup-coder/mcp4coder.git
cd mcp4coder

# 安装项目（含开发依赖）
pip install -e ".[dev]"

# 可选：安装特定 Server 的依赖
pip install -e ".[browser]"     # 浏览器自动化
pip install -e ".[google]"      # Google Sheets
pip install -e ".[notion]"      # Notion
pip install -e ".[database]"    # 数据库驱动
pip install -e ".[all]"         # 全部可选依赖

cp config/.env.example .env
# 编辑 .env 填入你的 API Key

# 启动 Redis（如果用 Docker）
docker compose -f deployment/docker-compose.yml up -d redis

# 启动应用
python -m app.main

# 另开终端：启动 Celery Worker
celery -A app.workers.celery_app worker --loglevel=info

# 开发环境 Demo Token
curl -X POST "http://localhost:8000/auth/token?username=demo&password=demo"
# → {"access_token": "eyJ...", "token_type": "bearer"}

curl http://localhost:8000/mcp/servers

AUTH="Authorization: Bearer <your_token>"

# 读取文件
curl -X POST http://localhost:8000/mcp/execute/filesystem/read_file \
  -H "$AUTH" -H "Content-Type: application/json" \
  -d '{"path": "README.md"}'

# 查询 Context7 文档
curl -X POST http://localhost:8000/mcp/execute/context7/query_docs \
  -H "$AUTH" -H "Content-Type: application/json" \
  -d '{"libraryId": "/facebook/react", "query": "useEffect cleanup"}'

# 结构化推理
curl -X POST http://localhost:8000/mcp/execute/sequential_thinking/sequential_thinking \
  -H "$AUTH" -H "Content-Type: application/json" \
  -d '{"thought": "分析问题：用户登录失败的可能原因...", "thoughtNumber": 1, "totalThoughts": 3, "nextThoughtNeeded": true}'

# 执行终端命令
curl -X POST http://localhost:8000/mcp/execute/desktop_commander/execute_command \
  -H "$AUTH" -H "Content-Type: application/json" \
  -d '{"command": "git log --oneline -5", "timeout_ms": 10000}'

# E2B 代码沙箱
curl -X POST http://localhost:8000/mcp/execute/e2b/execute_code \
  -H "$AUTH" -H "Content-Type: application/json" \
  -d '{"code": "print(sum(range(100)))", "language": "python"}'

# Sentry 查看 Issue
curl -X POST http://localhost:8000/mcp/execute/sentry/get_issue \
  -H "$AUTH" -H "Content-Type: application/json" \
  -d '{"issue_id_or_url": "123456"}'

# GitHub 搜索代码
curl -X POST http://localhost:8000/mcp/execute/github/search_code \
  -H "$AUTH" -H "Content-Type: application/json" \
  -d '{"q": "function main language:python"}'

# 触发通勤检查
curl -X POST http://localhost:8000/commute/check -H "$AUTH"
# → {"task_id": "abc-123", "status": "started"}

# 查询任务状态
curl http://localhost:8000/commute/status/abc-123

docker compose -f deployment/docker-compose.yml up -d

docker compose -f deployment/docker-compose.prod.yml up -d

docker build -f docker/Dockerfile -t mcp4coder:latest .
docker run -p 8000:8000 --env-file .env mcp4coder:latest

mcp4coder/
├── app/
│   ├── main.py                         # FastAPI 入口 + REST API 路由
│   ├── config/
│   │   ├── settings.py                 # Pydantic Settings 配置管理
│   │   └── validators.py              # 配置校验器
│   ├── mcp/
│   │   ├── servers/                    # 26 个 MCP Server 实现
│   │   │   ├── base_server.py          # BaseMCPServer 抽象基类
│   │   │   ├── amap_server.py          # 高德地图
│   │   │   ├── dingtalk_server.py      # 钉钉
│   │   │   ├── weather_server.py       # 天气
│   │   │   ├── calendar_server.py      # 日历
│   │   │   ├── filesystem_server.py    # 文件系统
│   │   │   ├── git_server.py           # Git
│   │   │   ├── database_server.py      # 数据库
│   │   │   ├── http_client_server.py   # HTTP 客户端
│   │   │   ├── github_server.py        # GitHub
│   │   │   ├── slack_server.py         # Slack
│   │   │   ├── brave_search_server.py  # Brave 搜索
│   │   │   ├── notion_server.py        # Notion
│   │   │   ├── google_sheets_server.py # Google 表格
│   │   │   ├── browser_server.py       # 浏览器自动化
│   │   │   ├── memory_server.py        # 记忆库
│   │   │   ├── context7_server.py      # Context7 文档注入
│   │   │   ├── sequential_thinking_server.py  # 结构化推理
│   │   │   ├── desktop_commander_server.py    # 终端命令
│   │   │   ├── docfork_server.py       # Docfork 文档搜索
│   │   │   ├── deepwiki_server.py      # DeepWiki 文档转换
│   │   │   ├── figma_server.py         # Figma 设计协作
│   │   │   ├── reactbits_server.py     # ReactBits 组件库
│   │   │   ├── e2b_server.py           # E2B 代码沙箱
│   │   │   ├── sentry_server.py        # Sentry 错误监控
│   │   │   ├── linear_server.py        # Linear 项目管理
│   │   │   └── composio_server.py      # Composio 多平台集成
│   │   ├── server_factory.py           # ServerType 枚举 + 工厂
│   │   └── server_manager.py           # 进程管理 + 健康检查
│   ├── middleware/
│   │   ├── auth.py                     # JWT + API Key 认证
│   │   └── security.py                 # 安全头 + 输入净化
│   ├── services/
│   │   └── commute_service.py          # 通勤助手业务逻辑
│   ├── workers/
│   │   ├── celery_app.py               # Celery 配置
│   │   └── tasks.py                    # 异步任务定义
│   ├── monitoring/
│   │   ├── performance_monitor.py      # 性能监控
│   │   └── system_monitor.py           # 系统监控
│   └── utils/
│       ├── logger.py                   # Loguru 日志
│       ├── error_handler.py            # 统一错误处理
│       └── helpers.py                  # 工具函数
├── web/
│   ├── mcp_manager.html                # Web 管理界面
│   └── workflow_designer.html          # 可视化工作流设计器
├── tests/
│   ├── conftest.py                     # 测试 fixtures
│   ├── test_config.py                  # 配置测试
│   ├── test_mcp.py                     # MCP Server 测试
│   ├── test_integration.py            # 集成测试
│   ├── test_e2e.py                     # 端到端测试
│   ├── test_security.py               # 安全测试
│   └── test_utils.py                   # 工具函数测试
├── docs/                               # 项目文档
│   ├── quick_start_guide.md
│   ├── detailed_operating_steps.md
│   ├── deployment.md
│   ├── SECURITY_CHECKLIST.md
│   └── ...
├── dumb_mode/                          # 极简独立模式（无需 FastAPI）
│   ├── commute_assistant.py
│   └── README_DUMB.md
├── docker/
│   └── Dockerfile
├── deployment/
│   ├── docker-compose.yml
│   └── docker-compose.prod.yml
├── scripts/
│   └── health_check.py
├── config/
│   ├── .env.example                    # 环境变量模板
│   ├── requirements.txt
│   └── pytest.ini
├── pyproject.toml                      # 项目元数据 + 工具配置
├── .bandit.yml                         # Bandit 安全扫描配置
├── .pre-commit-config.yaml             # Pre-commit hooks
├── CHANGELOG.md
├── CONTRIBUTING.md
└── CODE_OF_CONDUCT.md

# app/mcp/servers/my_server.py
from .base_server import BaseMCPServer, MCPTool, MCPResource, ServerCapability
from typing import Any, Dict, Optional

class MYMCPServer(BaseMCPServer):
    """命名规则：{VALUE}MCPServer，与 ServerType 枚举值对应"""

    def __init__(self, config: Optional[Dict[str, Any]] = None):
        # 在 super().__init__ 之前提取配置
        self.api_key = config.get("api_key", "") if config else ""
        super().__init__(config)

    def register_tools(self) -> None:
        self._register_tool(MCPTool(
            name="my_tool",
            description="工具描述",
            input_schema={
                "type": "object",
                "properties": {
                    "param1": {"type": "string", "description": "参数说明"}
                },
                "required": ["param1"]
            },
            capability=ServerCapability.READ  # READ | WRITE | EXECUTE
        ))

    def register_resources(self) -> None:
        self._register_resource(MCPResource(
            uri="my_server://status",
            name="状态",
            description="当前状态"
        ))

    async def execute_tool(self, tool_name: str, params: Dict[str, Any]) -> Any:
        if tool_name == "my_tool":
            return {"result": f"处理 {params['param1']}"}
        raise ValueError(f"未知工具: {tool_name}")

    async def _read_resource_content(self, resource: MCPResource) -> Any:
        if resource.uri == "my_server://status":
            return {"status": "ok"}
        raise ValueError(f"未知资源: {resource.uri}")

# ServerType 枚举
class ServerType(Enum):
    # ... 已有枚举
    MY = "my"  # 新增

# _load_builtin_servers()
ServerType.MY: ManagedServer(
    name="my-mcp-server",
    command="python",
    args=["-m", "app.mcp.servers.my_server"],
    env={"PYTHONPATH": "."},
    working_dir=".",
    timeout=300,
    max_concurrent=10,
    auto_restart=True,
    health_check_interval=120
),

# 语法检查
python3 -c "import ast; ast.parse(open('app/mcp/servers/my_server.py').read())"

# 运行测试
pytest tests/ -v

cd dumb_mode
pip install -r requirements.txt
python commute_assistant.py

# 安装 pre-commit hooks
pre-commit install

# 运行完整检查
black app/ && isort app/ && flake8 app/ && mypy app/ && pytest tests/