前景预测-api-mcp
项目介绍
Prospectio MCP API
基于 FastAPI 的应用程序，实现了用于潜在客户挖掘的模型上下文协议（MCP）。该项目遵循干净架构原则，在领域层、应用层和基础设施层之间有明确的关注点分离。
该应用程序现在包括持久存储能力，与 PostgreSQL 和 pgvector 集成，允许高效地存储和管理潜在客户数据。
🏗️ 项目架构
此项目实现 干净架构（也称为六边形架构），具有以下层次：
- 领域层：核心业务实体和逻辑
- 应用层：用例和 API 路由
- 基础设施层：外部服务、API 和框架实现
📁 项目结构
prospectio-api-mcp/
├── Dockerfile
├── README.md
├── curls/
│ └── list.http
├── database/
│ └── init.sql
├── docker-compose.yml
├── glama.json
├── poetry.lock
├── prospectio_api_mcp/
│ ├── __pycache__/
│ ├── application/
│ │ ├── api/
│ │ │ ├── leads_routes.py
│ │ │ ├── mcp_routes.py
│ │ │ ├── profile_routes.py
│ │ │ └── __pycache__/
│ │ └── use_cases/
│ │ ├── get_leads.py
│ │ ├── insert_leads.py
│ │ ├── profile.py
│ │ └── __pycache__/
│ ├── config.py
│ ├── domain/
│ │ ├── entities/
│ │ │ ├── company.py
│ │ │ ├── compatibility_score.py
│ │ │ ├── contact.py
│ │ │ ├── job.py
│ │ │ ├── leads.py
│ │ │ ├── leads_result.py
│ │ │ ├── profile.py
│ │ │ ├── work_experience.py
│ │ │ └── __pycache__/
│ │ ├── ports/
│ │ │ ├── compatibility_score.py
│ │ │ ├── fetch_leads.py
│ │ │ ├── leads_repository.py
│ │ │ ├── profile_respository.py
│ │ │ └── __pycache__/
│ │ ├── prompts/
│ │ │ └── compatibility_score.md
│ │ └── services/
│ │ ├── prompt_loader.py
│ │ ├── __pycache__/
│ │ └── leads/
│ │ ├── active_jobs_db.py
│ │ ├── jsearch.py
│ │ ├── mantiks.py
│ │ └── strategy.py
│ ├── infrastructure/
│ │ ├── api/
│ │ │ ├── client.py
│ │ │ ├── llm_client_factory.py
│ │ │ ├── llm_generic_client.py
│ │ │ └── __pycache__/
│ │ ├── dto/
│ │ │ ├── database/
│ │ │ ├── llm/
│ │ │ ├── mantiks/
│ │ │ └── rapidapi/
│ │ └── services/
│ │ ├── active_jobs_db.py
│ │ ├── compatibility_score.py
│ │ ├── jsearch.py
│ │ ├── leads_database.py
│ │ ├── mantiks.py
│ │ └── profile_database.py
│ ├── main.py
│ ├── mcp.py
│ ├── mcp_routes.py
│ └── __pycache__/
├── pyproject.toml
├── pyrightconfig.json
├── tests/
│ └── ut/
│ ├── test_1_profile_use_case.py
│ ├── test_active_jobs_db_use_case.py
│ ├── test_get_leads_use_case.py
│ ├── test_jsearch_use_case.py
│ ├── test_mantiks_use_case.py
│ └── __pycache__/
├── uv.lock
🔧 核心组件
领域层 (prospectio_api_mcp/domain/)
实体
Contact(contact.py)：表示一个商业联系人（姓名、电子邮件、电话、职位）Company(company.py)：表示一家公司（名称、行业、规模、位置、描述）Job(job.py)：表示一份工作招聘（职位、描述、地点、薪酬、要求）Leads(leads.py)：聚合公司、工作和联系人以形成潜在客户数据LeadsResult(leads_result.py)：表示插入潜在客户的操作结果Profile(profile.py)：表示带有个人和职业信息的用户档案WorkExperience(work_experience.py)：表示档案中的工作经验条目
端口
CompanyJobsPort(fetch_leads.py)：从任何数据源获取公司工作的抽象接口fetch_company_jobs(location: str, job_title: list[str]) -> Leads：抽象方法，用于工作搜索
LeadsRepositoryPort(leads_repository.py)：持久化潜在客户数据的抽象接口save_leads(leads: Leads) -> None：抽象方法，用于将潜在客户保存到存储中
ProfileRepositoryPort(profile_respository.py)：档案数据管理的抽象接口- 档案相关的仓库操作
策略 (prospectio_api_mcp/domain/services/leads/)
CompanyJobsStrategy(strategy.py)：工作检索策略的抽象基类- 具体策略：每个数据源的实现：
ActiveJobsDBStrategy,JsearchStrategy,MantiksStrategy
应用层 (prospectio_api_mcp/application/)
API (prospectio_api_mcp/application/api/)
leads_routes.py：定义用于潜在客户管理的 FastAPI 端点profile_routes.py：定义用于档案管理的 FastAPI 端点
用例 (prospectio_api_mcp/application/use_cases/)
InsertCompanyJobsUseCase(insert_leads.py)：协调从不同来源检索并插入公司工作的过程- 接受一个策略和仓库，检索潜在客户并将它们持久化到数据库中
GetLeadsUseCase(get_leads.py)：处理潜在客户数据的检索ProfileUseCase(profile.py)：管理与档案相关的操作
基础设施层 (prospectio_api_mcp/infrastructure/)
API 客户端 (prospectio_api_mcp/infrastructure/api/client.py)
BaseApiClient：用于外部 API 调用的异步 HTTP 客户端
DTOs (prospectio_api_mcp/infrastructure/dto/)
- 数据库 DTOs：
base.py,company.py,job.py,contact.py,profile.py,work_experience.py- 用于持久化的 SQLAlchemy 模型 - Mantiks DTOs：
company.py,company_response.py,job.py,location.py,salary.py- Mantiks API 的数据传输对象 - RapidAPI DTOs：
active_jobs_db.py,jsearch.py- RapidAPI 服务的数据传输对象
服务 (prospectio_api_mcp/infrastructure/services/)
ActiveJobsDBAPI：Active Jobs DB API 的适配器JsearchAPI：Jsearch API 的适配器MantiksAPI：Mantiks API 的适配器LeadsDatabase：用于潜在客户持久化的 PostgreSQL 仓库实现ProfileDatabase：用于档案管理的 PostgreSQL 仓库实现
所有 API 服务都实现了 CompanyJobsPort 接口，数据库服务实现了 LeadsRepositoryPort 接口，这使得替换和扩展变得容易。
🚀 应用程序入口点 (prospectio_api_mcp/main.py)
FastAPI 应用程序配置为：
- 管理应用程序生命周期：处理启动和关闭事件，包括 MCP 会话生命周期。
- 暴露多种协议：
- 在
/rest/v1/可用的 REST API - 在
/prospectio/可用的 MCP 协议（在mcp_routes.py中实现）
- 在
- 集成路由器：包括潜在客户插入路由和档案路由，通过 FastAPI 的 APIRouter 进行全面的潜在客户和档案管理。
- 加载配置：使用 Pydantic 从
config.py加载环境设置。 - 依赖注入：将服务实现、策略和仓库注入端点，实现干净的分离。
- 数据库集成：配置 PostgreSQL 连接，用于持久存储潜在客户数据和档案。
⚙️ 配置
要运行应用程序，您需要配置您的环境变量。这是通过在项目根目录创建一个 .env 文件来完成的。
-
创建
.env文件： 复制示例文件.env.example到一个新文件名为.env。cp .env.example .env cp .env .env.docker -
编辑
.env文件： 打开.env文件，并填写以下变量所需的值：EXPOSE：stdio或httpMASTER_KEY：您的主密钥。ALLOWED_ORIGINS：允许的来源列表，逗号分隔。MANTIKS_API_URL：Mantiks API 的基础 URL。MANTIKS_API_KEY：您的 Mantiks API 密钥。RAPIDAPI_API_KEY：您的 RapidAPI 密钥。JSEARCH_API_URL：Jsearch API 的基础 URL。ACTIVE_JOBS_DB_URL：Active Jobs DB API 的基础 URL。DATABASE_URL：PostgreSQL 连接字符串（例如，postgresql+asyncpg://user:password@host:port/database）
应用程序使用 Pydantic 设置从 .env 文件加载这些变量（参见 prospectio_api_mcp/config.py）。
📦 依赖项 (pyproject.toml)
核心依赖项
- FastAPI (0.115.14)：具有自动 API 文档的现代 Web 框架
- MCP (1.10.1)：模型上下文协议实现
- Pydantic (2.10.3)：数据验证和序列化
- HTTPX (0.28.1)：用于外部 API 调用的 HTTP 客户端
- SQLAlchemy (2.0.41)：用于 PostgreSQL 集成的数据库 ORM
- asyncpg (0.30.0)：异步 PostgreSQL 驱动
- psycopg (3.2.4)：PostgreSQL 适配器
开发依赖项
- Pytest：测试框架
🔄 数据流
- HTTP 请求：客户端向
/rest/v1/insert/leads/{source}发送带有 JSON 正文的 POST 请求，其中包含位置和职位参数。 - 路由处理器：FastAPI 路由在
application/api/routes.py接收请求并提取参数。 - 策略映射：处理器根据来源选择适当的策略（例如，
ActiveJobsDBStrategy,JsearchStrategy等）。 - 用例执行：实例化
InsertCompanyJobsUseCase并传入选定的策略和仓库。 - 策略执行：用例委托给策略的
execute()方法以获取潜在客户数据。 - 端口执行：策略调用端口的
fetch_company_jobs(location, job_title)方法，该方法由基础设施适配器（例如，ActiveJobsDBAPI）实现。
🧪 测试
该项目包括遵循 pytest 最佳实践和干净架构原则的综合单元测试。测试位于 tests/ 目录中，并使用依赖注入来模拟外部服务。
测试结构
tests/
└── ut/ # 单元测试
├── test_mantiks_use_case.py # Mantiks 策略测试
├── test_jsearch_use_case.py # JSearch 策略测试
├── test_active_jobs_db_use_case.py # Active Jobs DB 策略测试
├── test_get_leads.py # 获取潜在客户用例测试
└── test_profile.py # 档案用例测试
运行测试
安装依赖项：
poetry install
运行所有测试：
# 运行所有测试
poetry run pytest
# 运行带有详细输出
poetry run pytest -v
运行特定测试文件：
# 仅运行 Mantiks 测试
poetry run pytest tests/ut/test_mantiks_use_case.py -v
# 仅运行 JSearch 测试
poetry run pytest tests/ut/test_jsearch_use_case.py -v
# 仅运行 Active Jobs DB 测试
poetry run pytest tests/ut/test_active_jobs_db_use_case.py -v
# 仅运行 获取潜在客户 测试
poetry run pytest tests/ut/test_get_leads.py -v
# 仅运行 档案 测试
poetry run pytest tests/ut/test_profile.py -v
运行特定测试方法：
# 运行特定测试方法
poetry run pytest tests/ut/test_mantiks_use_case.py::TestMantiksUseCase::test_get_leads_success -v
测试环境变量
测试需要一个 .env 文件进行配置。复制示例文件：
cp .env.example .env
CI 管道自动处理环境设置和数据库初始化。
🏃♂️ 运行应用程序
在运行应用程序之前，请确保按照 配置 部分所述设置好您的环境变量。
选项 1：本地开发
-
安装依赖项：
poetry install -
运行应用程序：
poetry run fastapi run prospectio_api_mcp/main.py --reload --port <YOUR_PORT>
选项 2：Docker Compose（推荐）
Docker Compose 设置包括应用程序和带有 pgvector 扩展的 PostgreSQL 数据库。
首先为 prospectio 创建一个网络：
docker network create prospectio
-
构建并运行 Docker Compose：
# 构建并启动容器 docker-compose up --build # 或者后台运行（分离模式） docker-compose up -d --build -
停止应用程序：
# 停止容器 docker-compose down # 停止并删除卷（如果需要） docker-compose down -v -
查看日志：
# 查看实时日志 docker-compose logs -f # 查看特定服务的日志 docker-compose logs -f prospectio-api-mcp
### 访问 API
一旦应用程序运行（本地或通过 Docker），您可以访问：
- **REST API**：`http://localhost:<YOUR_PORT>/rest/v1/insert/leads/{source}`
- `source` 可以是：mantiks, active_jobs_db, jsearch
- 方法：POST，带有包含 `location` 和 `job_title` 数组的 JSON 正文
- 示例：`http://localhost:<YOUR_PORT>/rest/v1/insert/leads/mantiks`
- **API 文档**：`http://localhost:<YOUR_PORT>/docs`
- **MCP 端点**：`http://localhost:<YOUR_PORT>/prospectio/mcp/sse`
# 添加到 claude
更改 settings json 以匹配您的环境
```json
{
"mcpServers": {
"Prospectio-stdio": {
"command": "<ABSOLUTE_PATH>/uv",
"args": [
"--directory",
"<PROJECT_ABSOLUTE_PATH>",
"run",
"prospectio_api_mcp/main.py"
]
}
}
}
添加到 Gemini cli
更改 settings json 以匹配您的环境
{
"mcpServers": {
"prospectio-http": {
"httpUrl": "http://localhost:<YOUR_PORT>/prospectio/mcp/sse",
"timeout": 30000
},
"Prospectio-stdio": {
"command": "<ABSOLUTE_PATH>/uv",
"args": [
"--directory",
"<PROJECT_ABSOLUTE_PATH>",
"run",
"prospectio_api_mcp/main.py"
]
}
}
}
由 Prospectio 团队精心打造 ❤️