Apcore MCP Python

apcore mcp

用于apcore的自动MCP服务器和OpenAI工具桥。

apcore mcp 转动任何 apcore-将项目转化为MCP服务器和OpenAI工具提供商 零代码更改 您现有的项目。

┌──────────────────┐
│  django-apcore   │  ← your existing apcore project (unchanged)
│  flask-apcore    │
│  ...             │
└────────┬─────────┘
         │  extensions directory
         ▼
┌──────────────────┐
│    apcore-mcp    │  ← just install & point to extensions dir
└───┬──────────┬───┘
    │          │
    ▼          ▼
  MCP       OpenAI
 Server      Tools

设计理念

• 零入侵 --您的apcore项目不需要更改代码，不需要导入，也不需要依赖apcore mcp
• 零配置 --指向扩展目录，所有内容都会自动发现
• 纯适配器 --apcore mcp从apcore注册表中读取；它永远不会修改你的模块
• 适用于任何 xxx-apcore 项目 --如果它使用apcore模块注册表，apcore mcp可以为其提供服务

文档

有关完整文档，包括Python和TypeScript的快速入门指南，请访问： ****

安装

在现有的apcore项目旁边安装apcore mcp：

pip install apcore-mcp

就是这样。您现有的项目不需要更改。

快速开始

现在试试

该仓库包含5个示例模块（基于类+绑定.yaml），您可以立即运行：

pip install -e .
PYTHONPATH=./examples/binding_demo python examples/run.py
# Open http://127.0.0.1:8000/explorer/

看 示例/README.md 查看所有运行模式和模块详细信息。

零代码方法（CLI）

如果你已经有一个基于apcore的带有扩展目录的项目，只需运行：

apcore-mcp --extensions-dir /path/to/your/extensions

所有模块都是自动发现的，并作为MCP工具公开。不需要代码。

编程方法（Python API）

这 APCoreMCP 类是推荐的入口点——一个对象，所有功能：

from apcore_mcp import APCoreMCP

mcp = APCoreMCP("./extensions")

# Launch as MCP Server
mcp.serve()

# Or with HTTP + Explorer UI
mcp.serve(transport="streamable-http", port=8000, explorer=True)

# Or export as OpenAI tools
tools = mcp.to_openai_tools()

您还可以传递现有的 Registry 或 Executor:

from apcore import Registry
from apcore_mcp import APCoreMCP

registry = Registry(extensions_dir="./extensions")
registry.discover()
mcp = APCoreMCP(registry, name="my-server", tags=["public"])

Function-based API (still supported)

from apcore import Registry
from apcore_mcp import serve, to_openai_tools

registry = Registry(extensions_dir="./extensions")
registry.discover()

serve(registry)
tools = to_openai_tools(registry)

与现有项目集成

典型apcore项目结构

your-project/
├── extensions/          ← modules live here
│   ├── image_resize/
│   ├── text_translate/
│   └── ...
├── your_app.py          ← your existing code (untouched)
└── ...

添加MCP支持

您的项目没有更改。只需在它旁边运行apcore mcp：

# Install (one time)
pip install apcore-mcp

# Run
apcore-mcp --extensions-dir ./extensions

您现有的应用程序继续像以前一样工作。apcore-mcp作为一个单独的进程运行，从同一个扩展目录读取。

添加OpenAI工具支持

对于OpenAI集成，需要一个精简的脚本，但仍然 不对现有模块进行更改:

from apcore import Registry
from apcore_mcp import to_openai_tools

registry = Registry(extensions_dir="./extensions")
registry.discover()

tools = to_openai_tools(registry)
# Use with openai.chat.completions.create(tools=tools)

MCP客户端配置

克劳德桌面

添加 ~/Library/Application Support/Claude/claude_desktop_config.json （macOS）或 %APPDATA%\Claude\claude_desktop_config.json （Windows）：

{
  "mcpServers": {
    "apcore": {
      "command": "apcore-mcp",
      "args": ["--extensions-dir", "/path/to/your/extensions"]
    }
  }
}

克劳德代码

添加 .mcp.json 在项目根目录中：

{
  "mcpServers": {
    "apcore": {
      "command": "apcore-mcp",
      "args": ["--extensions-dir", "./extensions"]
    }
  }
}

光标

添加 .cursor/mcp.json 在项目根目录中：

{
  "mcpServers": {
    "apcore": {
      "command": "apcore-mcp",
      "args": ["--extensions-dir", "./extensions"]
    }
  }
}

远程HTTP访问

apcore-mcp --extensions-dir ./extensions \
    --transport streamable-http \
    --host 0.0.0.0 \
    --port 9000

将任何MCP客户端连接到 http://your-host:9000/mcp.

CLI参考

apcore-mcp --extensions-dir PATH [OPTIONS]

JWT密钥解析优先级： --jwt-key-file > --jwt-secret > APCORE_JWT_SECRET 环境变量。

退出代码： 0 正常， 1 无效参数， 2 启动失败。

Python API参考

APCoreMCP （推荐）

统一入口点——配置一次，随处使用：

from apcore_mcp import APCoreMCP

mcp = APCoreMCP(
    "./extensions",              # path, Registry, or Executor
    name="apcore-mcp",          # server name
    version=None,                # defaults to package version
    tags=None,                   # filter modules by tags
    prefix=None,                 # filter modules by ID prefix
    log_level=None,              # logging level ("DEBUG", "INFO", etc.)
    validate_inputs=False,       # validate inputs against schemas
    metrics_collector=None,      # MetricsExporter | bool — `True` auto-instantiates the collector
    observability=False,         # enable MetricsMiddleware + UsageMiddleware + /metrics + /api/usage
    authenticator=None,          # Authenticator for JWT/token auth (HTTP only)
    require_auth=True,           # False = permissive mode (no 401)
    exempt_paths=None,           # exact paths that bypass auth
    approval_handler=None,       # approval handler for runtime approval
    output_formatter=None,        # default: None (raw JSON); pass to_markdown to opt into apcore-toolkit Markdown
    middleware=None,             # list[Middleware] — user middleware applied after built-ins
    acl=None,                    # apcore.ACL — module access control
    async_tasks=True,            # enable F-043 Async Task Bridge
    async_max_concurrent=10,     # max concurrent async tasks
    async_max_tasks=1000,        # max queued async tasks
)

# Note: redact_output, strategy, and trace are configurable on the
# function-based serve() / async_serve(); they are not exposed on
# APCoreMCP.serve() (see the `serve()` reference below).

# Launch as MCP server (blocking)
mcp.serve(transport="streamable-http", port=8000, explorer=True)

# Export as OpenAI tools
tools = mcp.to_openai_tools(strict=True)

# Embed into ASGI app
async with mcp.async_serve(explorer=True) as app:
    ...

# Inspect
mcp.tools       # list of module IDs
mcp.registry    # underlying Registry
mcp.executor    # underlying Executor

serve() （基于功能）

from apcore_mcp import serve

serve(
    registry_or_executor,        # Registry or Executor
    transport="stdio",           # "stdio" | "streamable-http" | "sse"
    host="127.0.0.1",           # host for HTTP transports
    port=8000,                   # port for HTTP transports
    name="apcore-mcp",          # server name
    version=None,                # defaults to package version
    on_startup=None,             # callback before transport starts
    on_shutdown=None,            # callback after transport completes
    tags=None,                   # filter modules by tags
    prefix=None,                 # filter modules by ID prefix
    log_level=None,              # logging level ("DEBUG", "INFO", etc.)
    dynamic=False,               # rebuild tools on registry events
    validate_inputs=False,       # validate inputs against schemas
    metrics_collector=None,      # MetricsExporter | bool — `True` auto-instantiates apcore.observability.MetricsCollector
    explorer=False,              # enable browser-based Tool Explorer UI
    explorer_prefix="/explorer", # URL prefix for the explorer
    allow_execute=False,         # allow tool execution from the explorer
    explorer_title="MCP Tool Explorer",
    explorer_project_name=None,
    explorer_project_url=None,
    authenticator=None,          # Authenticator for JWT/token auth (HTTP only)
    require_auth=True,           # False = permissive mode (no 401)
    exempt_paths=None,           # exact paths that bypass auth
    approval_handler=None,       # approval handler for runtime approval
    output_formatter=None,       # default None (raw JSON); pass apcore_toolkit.to_markdown to opt in
    strategy=None,               # pipeline strategy preset: "standard" | "internal" | "testing" | "performance" | "minimal"
    redact_output=True,          # mask x-sensitive / _secret_* fields in outputs
    trace=False,                 # enable per-call apcore pipeline trace metadata
    middleware=None,             # list[Middleware] — applied after built-ins
    acl=None,                    # apcore.ACL — module access control
    observability=False,         # enable MetricsMiddleware + UsageMiddleware + /metrics + /api/usage
    async_tasks=True,            # enable F-043 Async Task Bridge
    async_max_concurrent=10,     # max concurrent async tasks
    async_max_tasks=1000,        # max queued async tasks
    # Note: schema_converter / annotation_mapper / error_mapper hooks are reserved for v0.16+ (EB-2)
)

接受a Registry 或 Executor.当a Registry 通过，a Executor 是自动创建的。

async_serve()

将MCP服务器嵌入到更大的ASGI应用程序中（例如与A2A、Django ASGI共同托管）：

from apcore_mcp import async_serve

async with async_serve(registry, explorer=True) as mcp_app:
    combined = Starlette(routes=[
        Mount("/mcp", app=mcp_app),
        Mount("/a2a", app=a2a_app),
    ])
    config = uvicorn.Config(combined, host="0.0.0.0", port=8000)
    await uvicorn.Server(config).serve()

接受与相同的参数 serve() （除 transport, host, port, on_startup, on_shutdown).返回a Starlette 应用程序通过异步上下文管理器。

工具资源管理器

当 explorer=True 传递给 serve()，基于浏览器的工具资源管理器UI安装在HTTP传输上。它提供了一个交互式页面，用于浏览工具模式和测试工具执行。

serve(registry, transport="streamable-http", explorer=True, allow_execute=True)
# Open http://127.0.0.1:8000/explorer/ in a browser

终点：

• 仅限HTTP传输 (streamable-http, sse).默默地忽略了 stdio.
• 默认情况下禁用执行 --set allow_execute=True 启用Try it。
• 自定义前缀 --使用 explorer_prefix="/browse" 以不同的路径安装。

JWT身份验证

HTTP传输的可选承载令牌身份验证。支持对称（HS256）和非对称（RS256）算法。

from apcore_mcp.auth import JWTAuthenticator

auth = JWTAuthenticator(key="my-secret")

serve(
    registry,
    transport="streamable-http",
    authenticator=auth,
    explorer=True,
    allow_execute=True,
)

允许模式 --允许未经身份验证的访问（身份为 None 当没有提供令牌时）：

serve(registry, transport="streamable-http", authenticator=auth, require_auth=False)

路径豁免 --绕过特定路径的身份验证：

serve(registry, transport="streamable-http", authenticator=auth, exempt_paths={"/health", "/metrics"})

看 示例/README.md 用于具有预生成测试令牌的可运行JWT演示。

审批机制

工具执行的可选运行时批准。将MCP启发与apcore的审批系统联系起来。

from apcore_mcp.adapters.approval import ElicitationApprovalHandler

handler = ElicitationApprovalHandler()

serve(
    registry,
    transport="streamable-http",
    approval_handler=handler,
    explorer=True,
)

内置处理程序：

CLI用法：

apcore-mcp --extensions-dir ./extensions --approval elicit

输出格式化

默认情况下，工具执行结果序列化为JSON(json.dumps).您可以通过传递 output_format 名称或自定义 output_formatter 可调用。

内置格式 （要求 apcore-toolkit 0.7+):

# Via CLI
# apcore-mcp --extensions-dir ./extensions --output-format csv

# Via API
mcp = APCoreMCP("./extensions", output_format="csv")

支持 json, csv，以及 jsonl非表格数据优雅地回落到JSON。

自定义格式化程序: 传递一个可调用函数，用于转换 dict 或 list 将结果转换为字符串。

def my_formatter(data: dict) -> str:
    return "\n".join(f"{k}: {v}" for k, v in data.items())

mcp = APCoreMCP("./extensions", output_formatter=my_formatter)

这 output_formatter 基于函数的参数也可用 serve() API及以上 ExecutionRouter 直接。

扩展助手

模块可以在执行过程中通过MCP协议回调报告进度并请求用户输入。当在MCP上下文之外调用时，这两个助手都不会优雅地执行操作。

from apcore_mcp import report_progress, elicit

# Inside a module's execute():
await report_progress(context, progress=50, total=100, message="Halfway done")

result = await elicit(context, "Confirm deletion?", {"type": "object", "properties": {"confirm": {"type": "boolean"}}})
if result and result["action"] == "accept":
    # proceed
    ...

/metrics 普罗米修斯端点

当 metrics_collector 提供给 serve()一 /metrics HTTP端点被公开，以Prometheus文本公开格式返回指标。

• 仅适用于基于HTTP的传输 (streamable-http, sse).不适用于 stdio 运输。
• 返回Prometheus文本格式 与内容类型 text/plain; version=0.0.4; charset=utf-8.
• 返回404 当否 metrics_collector 已配置。

from apcore.observability import MetricsCollector
from apcore_mcp import serve

collector = MetricsCollector()
serve(registry, transport="streamable-http", metrics_collector=collector)
# GET http://127.0.0.1:8000/metrics -> Prometheus text format

to_openai_tools()

from apcore_mcp import to_openai_tools

tools = to_openai_tools(
    registry_or_executor,       # Registry or Executor
    embed_annotations=False,    # append annotation hints to descriptions
    strict=False,               # OpenAI Structured Outputs strict mode
    tags=None,                  # filter by tags, e.g. ["image"]
    prefix=None,                # filter by module ID prefix, e.g. "image"
)

返回可直接用于OpenAI API的dicts列表：

import openai

client = openai.OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Resize the image to 512x512"}],
    tools=tools,
)

严格模式 (strict=True)：组 additionalProperties: false，使所有属性都是必需的（可选属性可以为空），删除默认值。

注释嵌入 (embed_annotations=True)：附加 [Annotations: read_only, idempotent] 描述。

过滤: tags=["image"] 或 prefix="text" 以暴露模块的子集。

与执行器一起使用

如果您需要自定义中间件、ACL或执行配置：

from apcore import Registry, Executor

registry = Registry(extensions_dir="./extensions")
registry.discover()
executor = Executor(registry)

serve(executor)
tools = to_openai_tools(executor)

特性

• 自动发现 --自动找到并公开扩展目录中的所有模块
• 显示叠加 — metadata["display"]["mcp"] 控制每个模块的MCP工具名称、描述和指导（§5.13）；通过设置 binding_path 在 fastapi-apcore
• Markdown工具说明 (rich_description=True，v0.15+）--渲染 Tool.description /OpenAI function.description 作为规范的apcore工具包Markdown（参数、返回、行为表、标签、示例），LLM每个令牌获得更多与决策相关的信号。
• 模块预览元工具 (__apcore_module_preview，v0.15+）——让AI编排器运行 executor.validate() 在不执行模块的情况下预测状态变化（apcore PROTOCOL_SPEC§5.6）。退货 {valid, requires_approval, predicted_changes, checks}.
• 三次运输 --stdio（默认，用于桌面客户端）、流式HTTP和SSE
• JWT身份验证 --HTTP传输的可选承载令牌身份验证 JWTAuthenticator、许可模式、PEM密钥文件支持和环境变量回退
• 审批机制 --通过MCP启发、自动批准或始终拒绝处理程序进行运行时批准
• AI指导 --错误响应包括 retryable, ai_guidance, user_fixable，以及 suggestion 代理消费字段
• AI意图元数据 --工具描述丰富 x-when-to-use, x-when-not-to-use, x-common-mistakes, x-workflow-hints 来自模块元数据
• 扩展助手 --模块可以调用 report_progress() 和 elicit() 在执行过程中，用于MCP进度报告和用户输入
• 注释映射 --apcore注释（只读、破坏性、幂等）映射到MCP工具注释
• 模式转换 --JSON模式 $ref/$defs OpenAI结构化输出的内联严格模式
• 错误清理 --ACL错误和内部错误被清除；堆栈痕迹永远不会泄漏
• 动态注册 --在运行时注册/未注册的模块会立即反映出来
• 双输出 --同一注册表为MCP服务器和OpenAI工具定义提供支持
• 工具资源管理器 --基于浏览器的UI，用于交互式浏览模式和测试工具，具有Swagger UI风格的身份验证输入
• 配置总线集成 --注册a mcp 带有apcore配置总线的命名空间；通过统一配置传输、主机、端口等 apcore.yaml 或 APCORE_MCP_* 环境变量
• 格式化程序注册表错误 --注册一个特定于MCP的错误格式化程序，用于全生态系统一致的错误处理

配置总线集成

apcore mcp注册了一个 mcp 在导入时使用apcore配置总线的命名空间。这意味着MCP设置可以与其他apcore配置一起使用 apcore.yaml:

apcore:
  version: "1.0.0"
mcp:
  transport: streamable-http
  host: 0.0.0.0
  port: 9000
  explorer: true
  require_auth: false

环境变量重写使用 APCORE_MCP_ 前缀：

APCORE_MCP_TRANSPORT=streamable-http
APCORE_MCP_PORT=9000
APCORE_MCP_EXPLORER=true

默认值： transport=stdio, host=127.0.0.1, port=8000, explorer=false, require_auth=true.

命名空间、前缀和默认值也可以作为可导入常量使用：

from apcore_mcp import MCP_NAMESPACE, MCP_ENV_PREFIX, MCP_DEFAULTS

运作原理

映射：apcore到MCP

映射：apcore到OpenAI工具

带有点的模块ID被标准化为破折号，以实现OpenAI兼容性（双射映射）。

建筑

Your apcore project (unchanged)
    │
    │  extensions directory
    ▼
apcore-mcp (separate process / library call)
    │
    ├── MCP Server path
    │     SchemaConverter + AnnotationMapper
    │       → MCPServerFactory → ExecutionRouter → TransportManager
    │
    └── OpenAI Tools path
          SchemaConverter + AnnotationMapper + IDNormalizer
            → OpenAIConverter → list[dict]

发展

git clone https://github.com/aiperceivable/apcore-mcp-python.git
cd apcore-mcp
pip install -e ".[dev]"
pytest                           # ~689 tests
pytest --cov                     # with coverage report

许可证

阿帕奇-2.0