AI模型集成协议服务器

Marketing Connect MCP Services 是一个基于Model Context Protocol (MCP)的服务器,用于简化AI模型与外部工具和数据源的集成,提供工具调用、资源加载和提示模板功能。

  • by alanedavis
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 alanedavis/mcp

使用说明

营销连接MCP服务

用于营销连接AI集成的模型上下文协议(MCP)服务器。

什么是MCP?

这 模型上下文协议(MCP) 是Anthropic的开放标准,使AI模型能够安全地与外部工具和数据源进行交互。此服务器公开:

  • 工具:AI可以调用的函数(如API端点)
  • 资源:加载到AI上下文中的数据(如配置或模式)
  • 鼓励:可重复使用的交互模板

什么是FastMCP?

FastMCP 是一个高级Python框架,简化了MCP服务器的构建。它提供了一个类似于FastAPI的基于装饰器的API,减少了样板文件并加速了开发。

为什么要使用FastMCP?

好处 说明
最小沸腾板 简单的装饰师喜欢 @mcp.tool() 替换复杂的类层次结构
自动模式生成 从Python类型提示生成的输入/输出模式
内置HTTP传输 具有健康检查和SSE支持的生产就绪服务器
Pydantic集成 原生支持Pydantic模型作为工具输入

MCP SDK与FastMCP的比较

无FastMCP (使用低级MCP SDK):

from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio

server = Server("my-server")

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="greet",
            description="Greet a user",
            inputSchema={
                "type": "object",
                "properties": {
                    "name": {"type": "string", "description": "User name"}
                },
                "required": ["name"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "greet":
        return [TextContent(type="text", text=f"Hello {arguments['name']}!")]

async def main():
    async with mcp.server.stdio.stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

使用FastMCP:

from fastmcp import FastMCP

mcp = FastMCP("my-server")

@mcp.tool()
async def greet(name: str) -> str:
    """Greet a user."""
    return f"Hello {name}!"

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

FastMCP将约30条线路减少到约10条,同时保持完全符合MCP协议。

快速开始

先决条件

从Devshell安装:

  • Python 3.11+ (3.13推荐)
  • make
  • buildi-cli
  • tfl
  • httpie

安装

# Install uv package manager
make ci-prebuild

# Install all dependencies (creates .venv automatically)
make build

模型生成

此服务器使用从OpenAPI规范生成的Pydantic模型。模型是通过以下方式生成的 datamodel-code-generator.

来自Local.tgz(npm打包的OpenAPI规范)

# Command to be run in marketing-connect-spec/marketing-connect-mcp-services (in terminal)
  # Inside path src/main/resources/model/api dir
tar -czvf models.tgz mcpservices.api.yml schema/

# Generate models from a local .tgz file
make generate-models SPEC_TGZ=path/to/models.tgz

来自工件URL

# Generate models from a URL (e.g., artifactory)
make generate-models-url SPEC_URL=https://artifactory.example.com/openapi-spec.tgz

来自本地YAML文件

# Generate models directly from a local OpenAPI YAML file
make generate-models-local SPEC_FILE=path/to/openapi.yaml

模型管理

# Show generated model classes
make models-show

# Clean generated models
make models-clean

生成的模型被放置在 src/marketing_connect_mcp_services/models/ 并且可以导入为:

from marketing_connect_mcp_services.models import ProductDetails, GreetingResponse

运行服务器

# Start the server (default: 0.0.0.0:8000)
make run

# Or with debug mode
make run-debug

# Or directly with uv
uv run marketing-connect-mcp --port 3000

验证部署

服务器公开健康检查端点以进行部署验证:

端点 描述
GET / 服务概述
GET /health 健康检查(退货 {"status": "UP"})
GET /info 服务器元数据(版本、配置、正常运行时间)
POST /mcp MCP协议端点(用于AI客户端)
# Check health
curl http://localhost:8000/health

# Get server info
curl http://localhost:8000/info

# Service overview
curl http://localhost:8000/

测试MCP协议

MCP端点使用Streamable HTTP传输,并需要特定的标头:

# Initialize MCP session
curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
      "protocolVersion": "2024-11-05",
      "capabilities": {},
      "clientInfo": {"name": "test-client", "version": "1.0"}
    }
  }'

预期响应(SSE格式):

event: message
data: {"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2024-11-05","capabilities":{...},"serverInfo":{"name":"marketing-connect-mcp-services","version":"..."}}}
# List available tools
curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "id": 2,
    "method": "tools/list",
    "params": {}
  }'

注: MCP协议是有状态的。这 initialize 请求在没有会话的情况下工作,但后续请求如下 tools/list 和 tools/call 需要会话ID标头(Mcp-Session-Id)从初始化响应中。对于完整的协议测试,请使用MCP客户端库

项目结构

marketing-connect-mcp-services/
├── src/marketing_connect_mcp_services/
│   ├── __init__.py          # Package exports
│   ├── server.py            # FastMCP server setup
│   ├── config.py            # Pydantic settings
│   ├── cli.py               # CLI entry point
│   ├── tools/               # MCP tools (AI-invokable functions)
│   │   ├── __init__.py
│   │   └── example.py       # Example tool patterns
│   ├── resources/           # MCP resources (context data)
│   │   ├── __init__.py
│   │   └── example.py       # Example resource patterns
│   └── prompts/             # MCP prompts (interaction templates)
│       ├── __init__.py
│       └── example.py       # Example prompt patterns
├── tests/                   # Test suite
├── pyproject.toml           # Hatchling build config + dependencies
├── uv.lock                  # Dependency lock file
├── Makefile                 # Build commands
└── .env.example             # Environment template

构建系统

本项目使用现代Python工具:

工具 目的
雏鸟 构建后端(PEP 517)
紫外线 快速包管理器(比pip快10-100倍)

为什么是紫外线?

  • 快速:用Rust编写,安装包比pip快10-100倍
  • 锁定文件: uv.lock 确保可重复构建
  • 兼容:符合标准 pyproject.toml
  • 简单:单一二进制文件,无需插件

配置

配置通过环境变量(前缀: MCP_).

复制 .env.example 到 .env 并自定义:

# Server identity
MCP_SERVER_NAME=marketing-connect-mcp-services
MCP_SERVER_VERSION=1.0.0

# HTTP server
MCP_HOST=0.0.0.0
MCP_PORT=8000

# Logging
MCP_DEBUG=false
MCP_LOG_LEVEL=INFO

# Application settings
MCP_BASE_URL=https://your-app-url.com
MCP_REGION=us-east-1

JPMC工件库

PyPI索引在中配置 pyproject.toml:

[tool.uv]
index-url = "https://artifacts-read.gkp.jpmchase.net/artifactory/api/pypi/pypi/simple"

您还可以通过环境变量进行覆盖:

export UV_INDEX_URL=https://your-pypi-mirror.com/simple

发展

测试

# Run tests
make test

# Run tests with coverage
make cover

# Verbose output
make test-verbose

代码质量

# Format code
make format

# Lint code
make lint

# Auto-fix lint issues
make lint-fix

# Type check
make typecheck

# Run all checks
make check

预提交钩子

make precommit

依赖管理

# Update lock file
make lock

# Update all dependencies to latest
make update

# Install production deps only
make build-prod

添加自定义集成

添加工具

在中创建新文件 tools/ 并注册它:

# tools/my_tools.py
from marketing_connect_mcp_services.server import mcp

@mcp.tool()
async def my_custom_tool(param: str) -> str:
    """Description the AI will see."""
    return f"Result: {param}"

然后导入 server.py:

from marketing_connect_mcp_services.tools import my_tools  # noqa: F401

添加资源

# resources/my_resources.py
from marketing_connect_mcp_services.server import mcp

@mcp.resource("myapp://config")
async def get_config() -> str:
    """Returns configuration data."""
    return "config data"

添加提示

# prompts/my_prompts.py
from marketing_connect_mcp_services.server import mcp

@mcp.prompt()
async def analysis_prompt(topic: str) -> str:
    """Generate an analysis prompt."""
    return f"Please analyze: {topic}"

CI/CD

# Full CI pipeline (clean, build, test, package)
make ci

# Generate SSAP reports
make ssap

# Build wheel package
make package

制定目标

目标 描述
make run 启动MCP服务器
make run-debug 从调试日志记录开始
make build 安装所有依赖项
make build-prod 仅安装生产设备
make test 运行测试
make cover 运行覆盖率测试
make format 格式代码
make lint Lint代码
make typecheck 运行mypy类型检查
make check 运行lint+类型检查+测试
make generate-models 从.tgz生成Pydantic模型
make generate-models-url 从URL生成模型
make generate-models-local 从本地YAML生成模型
make models-show 显示生成的模型类
make models-clean 删除生成的模型
make lock 更新uv.lock
make update 更新所有依赖项
make ci 完整的CI管道
make ssap 生成SSAP报告
make package 构建车轮
make help 显示所有目标

文档