使用说明
营销连接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推荐)makebuildi-clitflhttpie
安装
# 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 |
显示所有目标 |
