FinanceNLP MCP：用于金融自然语言处理的模型上下文协议服务器

![Python 3.8+](https://www.python.org/downloads/) ![FastAPI](https://fastapi.tiangolo.com/) ![License: MIT](https://opensource.org/licenses/MIT) ![MCP Compliant](https://modelcontextprotocol.io/)

一个生产就绪的模型上下文协议（MCP）服务器，支持标准化的大型语言模型与模块化金融NLP工具的集成。专为解析、分析和提取表格和非结构化财务数据中的见解而构建。

🎯 概述

FinanceNLP MCP通过提供标准化、可扩展的后端来支持特定于金融领域的多个NLP任务，从而弥合了LLM和金融数据处理之间的差距。无论您是在处理盈利报告、市场分析、监管文件还是实时金融新闻，该服务器都为您提供了所需的工具，并提供了一致的API界面。

关键价值主张

🔗 LLM集成：符合MCP的架构，可与语言模型无缝集成
📈 金融领域专业知识：专门为金融内容优化的NLP工具
🔄 多格式支持：处理表格数据、非结构化文本、JSON等
⚡ 生产就绪：使用FastAPI构建，用于高性能、可扩展的部署
🛠️ 可扩展设计：模块化架构，便于定制和添加工具

🏗️ 建筑

模型上下文协议（MCP）合规性

此服务器实现了模型上下文协议规范，提供：

标准化请求/响应格式：用于LLM集成的一致的API合同
工具发现：动态工具注册和能力广告
错误处理：具有详细元数据的强大错误报告
异步处理：非阻塞请求处理，可扩展性能

核心组件

graph TD
    A[LLM Client] --> B[MCP Server]
    B --> C[Tool Router]
    C --> D[Financial NLP Processor]
    D --> E[Summarization Engine]
    D --> F[Classification Engine]
    D --> G[Entity Extraction Engine]
    D --> H[Sentiment Analysis Engine]
    I[Market Data API] --> D
    J[External NLP Services] --> D

🚀 快速开始

先决条件

Python 3.8或更高版本
pip包管理器
互联网连接（用于市场数据和NLTK下载）

安装

克隆存储库

   git clone https://github.com/yourusername/FinanceNLP-MCP.git
   cd FinanceNLP-MCP

创建虚拟环境

   python -m venv finance-nlp-env
   source finance-nlp-env/bin/activate  # On Windows: finance-nlp-env\Scripts\activate

安装依赖项

   pip install -r requirements.txt

运行服务器

   python main.py

服务器将于启动 http://localhost:8000

Docker部署

# Build the Docker image
docker build -t finance-nlp-mcp .

# Run the container
docker run -p 8000:8000 finance-nlp-mcp

📚 API文档

核心MCP端点

`POST /mcp/process`

处理所有金融NLP任务的主要MCP处理端点。

请求格式：

{
    "tool_type": "summarization|classification|extraction|sentiment",
    "data_format": "text|json|tabular|unstructured",
    "input_data": "your_data_here",
    "parameters": {
        "max_length": 150,
        "custom_param": "value"
    },
    "context": "optional_context_information"
}

响应格式：

{
    "success": true,
    "tool_type": "summarization",
    "result": {
        "summary": "Generated summary text...",
        "key_sentences": 3,
        "confidence": 0.85
    },
    "metadata": {
        "data_format": "text",
        "processing_time": "2024-01-15T10:30:00Z",
        "parameters_used": {"max_length": 150}
    },
    "timestamp": "2024-01-15T10:30:00Z"
}

工具特定API

1.财务文本摘要

端点： POST /examples/summarize

使用针对财务内容优化的提取式摘要生成财务文档的简明摘要。

特征：

金融关键词权重
基于位置的句子评分
可配置摘要长度
信心评分

例子：

curl -X POST "http://localhost:8000/examples/summarize" \
  -H "Content-Type: application/json" \
  -d '{"text": "Apple Inc. reported record quarterly earnings of $123.9 billion in revenue, beating analyst expectations by 5%. The company saw strong growth in iPhone sales and services revenue, with CEO Tim Cook noting exceptional performance in international markets."}'

答复：

{
    "success": true,
    "result": {
        "summary": "Apple Inc. reported record quarterly earnings of $123.9 billion in revenue, beating analyst expectations by 5%. The company saw strong growth in iPhone sales and services revenue.",
        "key_sentences": 2,
        "confidence": 0.92
    }
}

2.财务文本分类

端点： POST /examples/classify

将财务文本分类到预定义的财务域中。

类别：

earnings_report：季度/年度收益公告
market_analysis：市场趋势和预测
company_news：公司公告和活动
regulatory：SEC文件和合规事项
economic_indicator：经济数据和指标

例子：

curl -X POST "http://localhost:8000/examples/classify" \
  -H "Content-Type: application/json" \
  -d '{"text": "The Federal Reserve announced a 0.25% interest rate increase to combat rising inflation, marking the third rate hike this year."}'

答复：

{
    "success": true,
    "result": {
        "primary_category": "economic_indicator",
        "confidence": 0.78,
        "all_scores": {
            "earnings_report": 0.0,
            "market_analysis": 0.2,
            "company_news": 0.1,
            "regulatory": 0.3,
            "economic_indicator": 0.78
        },
        "is_financial": true
    }
}

3.金融实体提取

端点： POST /examples/extract

从非结构化文本中提取结构化金融实体。

提取实体：

股票代码（例如AAPL、GOOGL）
货币金额（例如12亿美元、5亿美元）
百分比（例如，15%、0.25%）
日期（例如，2024年第三季度、2024年1月15日）
货币代码（美元、欧元、英镑等）

例子：

curl -X POST "http://localhost:8000/examples/extract" \
  -H "Content-Type: application/json" \
  -d '{"text": "AAPL stock rose 3.5% after reporting $89.5 billion in revenue for Q1 2024, with strong performance in the EUR and USD markets."}'

答复：

{
    "success": true,
    "result": {
        "companies": [],
        "currencies": ["EUR", "USD"],
        "amounts": ["$89.5 billion"],
        "dates": ["Q1 2024"],
        "percentages": ["3.5%"],
        "stock_symbols": ["AAPL"]
    }
}

4.金融情绪分析

端点： POST /examples/sentiment

将一般情绪与金融领域特定指标相结合的多层情绪分析。

分析层：

VADER情绪分析
文本Blob极性/主观性
金融关键词权重
综合信心评分

例子：

curl -X POST "http://localhost:8000/examples/sentiment" \
  -H "Content-Type: application/json" \
  -d '{"text": "The company exceeded profit expectations with strong growth in all segments, driving bullish investor sentiment."}'

答复：

{
    "success": true,
    "result": {
        "overall_sentiment": "positive",
        "confidence": 0.87,
        "scores": {
            "vader": {
                "compound": 0.6696,
                "pos": 0.294,
                "neu": 0.706,
                "neg": 0.0
            },
            "textblob": {
                "polarity": 0.5,
                "subjectivity": 0.75
            },
            "financial_context": 0.125,
            "combined": 0.431
        }
    }
}

市场数据集成

`GET /market/quote/{symbol}`

检索实时市场数据以进行财务分析。

例子：

curl "http://localhost:8000/market/quote/AAPL"

答复：

{
    "symbol": "AAPL",
    "current_price": 175.43,
    "company_name": "Apple Inc.",
    "market_cap": 2847234000000,
    "pe_ratio": 28.15,
    "timestamp": "2024-01-15T15:30:00Z"
}

系统端点

`GET /`

服务器信息和功能

`GET /health`

用于监控的健康检查端点

`GET /tools`

列出所有可用工具及其规格

🔧 配置

环境变量

创建一个 .env 项目根目录中的文件：

# Server Configuration
HOST=0.0.0.0
PORT=8000
LOG_LEVEL=info
WORKERS=1

# API Keys (if using external services)
ALPHA_VANTAGE_KEY=your_alpha_vantage_key
FINNHUB_KEY=your_finnhub_key

# Processing Configuration
MAX_SUMMARY_LENGTH=200
DEFAULT_CONFIDENCE_THRESHOLD=0.7

# Cache Configuration
REDIS_URL=redis://localhost:6379
CACHE_TTL=3600

自定义配置

修改 FinancialNLPProcessor 要自定义的类：

金融关键词词典
情绪分析权重
分类类别
实体提取模式

自定义示例：

class CustomFinancialProcessor(FinancialNLPProcessor):
    def __init__(self):
        super().__init__()
        self.financial_keywords.update({
            'crypto': ['bitcoin', 'ethereum', 'blockchain', 'cryptocurrency'],
            'esg': ['sustainability', 'carbon', 'green', 'environmental']
        })

🧪 测试

单元测试

# Run all tests
pytest tests/

# Run with coverage
pytest --cov=src tests/

# Run specific test categories
pytest tests/test_summarization.py
pytest tests/test_classification.py
pytest tests/test_extraction.py
pytest tests/test_sentiment.py

集成测试

# Test MCP compliance
pytest tests/test_mcp_compliance.py

# Test API endpoints
pytest tests/test_api_endpoints.py

# Performance tests
pytest tests/test_performance.py

手动测试

使用API交互式文档，网址为 http://localhost:8000/docs 手动测试端点。

示例测试用例：

收益报告处理：

   {
       "tool_type": "summarization",
       "data_format": "text",
       "input_data": "Microsoft Corporation today announced the following results for the quarter ended December 31, 2023, as compared to the corresponding period of last fiscal year: Revenue was $62.0 billion and increased 20% (up 19% in constant currency). Operating income was $27.0 billion and increased 23% (up 22% in constant currency). Net income was $22.3 billion and increased 33% (up 32% in constant currency). Diluted earnings per share was $2.93 and increased 33% (up 32% in constant currency)."
   }

市场分析分类：

   {
       "tool_type": "classification",
       "data_format": "text",
       "input_data": "Technical analysis suggests the S&P 500 is approaching a key resistance level at 4,800 points. Trading volume has been declining, and the RSI indicator shows overbought conditions. Analysts recommend caution in the near term."
   }

🔌 集成示例

Python客户端

import asyncio
import aiohttp

class FinanceNLPClient:
    def __init__(self, base_url="http://localhost:8000"):
        self.base_url = base_url
    
    async def process_text(self, text, tool_type="summarization", **kwargs):
        async with aiohttp.ClientSession() as session:
            payload = {
                "tool_type": tool_type,
                "data_format": "text",
                "input_data": text,
                "parameters": kwargs
            }
            
            async with session.post(
                f"{self.base_url}/mcp/process",
                json=payload
            ) as response:
                return await response.json()

# Usage
async def main():
    client = FinanceNLPClient()
    
    # Summarize financial text
    result = await client.process_text(
        "Apple reported strong quarterly results...",
        tool_type="summarization",
        max_length=100
    )
    print(result)

asyncio.run(main())

JavaScript/TypeScript客户端

interface MCPRequest {
    tool_type: 'summarization' | 'classification' | 'extraction' | 'sentiment';
    data_format: 'text' | 'json' | 'tabular' | 'unstructured';
    input_data: string | object | Array;
    parameters?: Record;
    context?: string;
}

class FinanceNLPClient {
    constructor(private baseUrl: string = 'http://localhost:8000') {}
    
    async processText(request: MCPRequest): Promise {
        const response = await fetch(`${this.baseUrl}/mcp/process`, {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json',
            },
            body: JSON.stringify(request),
        });
        
        return response.json();
    }
}

// Usage
const client = new FinanceNLPClient();

const result = await client.processText({
    tool_type: 'sentiment',
    data_format: 'text',
    input_data: 'The market showed strong bullish momentum today.',
});

console.log(result);

LangChain集成

from langchain.tools import Tool
from langchain.agents import AgentExecutor, create_openai_functions_agent

def create_finance_nlp_tools(client):
    """Create LangChain tools from FinanceNLP-MCP client"""
    
    def summarize_financial_text(text: str) -> str:
        result = asyncio.run(client.process_text(text, "summarization"))
        return result['result']['summary']
    
    def analyze_financial_sentiment(text: str) -> str:
        result = asyncio.run(client.process_text(text, "sentiment"))
        sentiment = result['result']['overall_sentiment']
        confidence = result['result']['confidence']
        return f"Sentiment: {sentiment} (confidence: {confidence:.2f})"
    
    return [
        Tool(
            name="summarize_financial_text",
            description="Summarize financial documents and reports",
            func=summarize_financial_text
        ),
        Tool(
            name="analyze_financial_sentiment",
            description="Analyze sentiment of financial text",
            func=analyze_financial_sentiment
        ),
    ]

📊 性能优化

缓存策略

对频繁访问的数据实施Redis缓存：

import redis
import json
from functools import wraps

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def cache_result(expiration=3600):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # Create cache key from function name and arguments
            cache_key = f"{func.__name__}:{hash(str(args) + str(kwargs))}"
            
            # Try to get cached result
            cached = redis_client.get(cache_key)
            if cached:
                return json.loads(cached)
            
            # Execute function and cache result
            result = await func(*args, **kwargs)
            redis_client.setex(
                cache_key, 
                expiration, 
                json.dumps(result, default=str)
            )
            
            return result
        return wrapper
    return decorator

批处理

在一个请求中处理多个文本：

@app.post("/mcp/batch")
async def batch_process(requests: List[MCPRequest]):
    """Process multiple MCP requests in batch"""
    tasks = [ToolRouter.route_request(req) for req in requests]
    results = await asyncio.gather(*tasks)
    return {"batch_results": results}

异步处理

对于长时间运行的任务，使用任务队列实现异步处理：

from celery import Celery

celery_app = Celery('finance_nlp')

@celery_app.task
def process_large_document(document_text, tool_type):
    """Process large documents asynchronously"""
    # Implementation here
    pass

@app.post("/mcp/async")
async def async_process(request: MCPRequest):
    """Submit async processing task"""
    task = process_large_document.delay(
        request.input_data, 
        request.tool_type
    )
    return {"task_id": task.id}

🔒 安全考虑

API安全

速率限制

   from slowapi import Limiter, _rate_limit_exceeded_handler
   from slowapi.util import get_remote_address

   limiter = Limiter(key_func=get_remote_address)
   app.state.limiter = limiter

   @app.post("/mcp/process")
   @limiter.limit("100/minute")
   async def process_mcp_request(request: Request, mcp_request: MCPRequest):
       # Implementation

输入验证

   from pydantic import validator

   class MCPRequest(BaseModel):
       # ... other fields ...
       
       @validator('input_data')
       def validate_input_size(cls, v):
           if isinstance(v, str) and len(v) > 100000:  # 100KB limit
               raise ValueError('Input data too large')
           return v

认证

   from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

   security = HTTPBearer()

   async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
       # Implement token verification
       pass

数据隐私

对敏感财务信息实施数据加密
添加请求/响应日志记录控件
提供数据保留策略
支持GDPR合规功能

🚀 部署

生产部署

Docker Compose

version: '3.8'

services:
  finance-nlp-mcp:
    build: .
    ports:
      - "8000:8000"
    environment:
      - WORKERS=4
      - LOG_LEVEL=info
    depends_on:
      - redis
      - postgres
    
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
  
  postgres:
    image: postgres:15-alpine
    environment:
      POSTGRES_DB: finance_nlp
      POSTGRES_USER: finance_user
      POSTGRES_PASSWORD: secure_password
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

Kubernetes部署

apiVersion: apps/v1
kind: Deployment
metadata:
  name: finance-nlp-mcp
spec:
  replicas: 3
  selector:
    matchLabels:
      app: finance-nlp-mcp
  template:
    metadata:
      labels:
        app: finance-nlp-mcp
    spec:
      containers:
      - name: finance-nlp-mcp
        image: finance-nlp-mcp:latest
        ports:
        - containerPort: 8000
        env:
        - name: WORKERS
          value: "2"
        - name: LOG_LEVEL
          value: "info"
        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "1Gi"
            cpu: "500m"
---
apiVersion: v1
kind: Service
metadata:
  name: finance-nlp-mcp-service
spec:
  selector:
    app: finance-nlp-mcp
  ports:
  - port: 80
    targetPort: 8000
  type: LoadBalancer

监测和可观察性

普罗米修斯指标

from prometheus_client import Counter, Histogram, generate_latest

REQUEST_COUNT = Counter('finance_nlp_requests_total', 'Total requests', ['tool_type', 'status'])
REQUEST_DURATION = Histogram('finance_nlp_request_duration_seconds', 'Request duration')

@app.middleware("http")
async def add_prometheus_metrics(request: Request, call_next):
    start_time = time.time()
    
    response = await call_next(request)
    
    duration = time.time() - start_time
    REQUEST_DURATION.observe(duration)
    REQUEST_COUNT.labels(
        tool_type=getattr(request.state, 'tool_type', 'unknown'),
        status=response.status_code
    ).inc()
    
    return response

@app.get("/metrics")
async def metrics():
    return Response(generate_latest(), media_type="text/plain")

健康检查

@app.get("/health/live")
async def liveness_check():
    """Kubernetes liveness probe"""
    return {"status": "alive"}

@app.get("/health/ready")
async def readiness_check():
    """Kubernetes readiness probe"""
    # Check dependencies
    try:
        # Test database connection
        # Test external API availability
        return {"status": "ready"}
    except Exception as e:
        raise HTTPException(status_code=503, detail="Not ready")

🤝 贡献

我们欢迎捐款！请查看我们的贡献指南了解详情。

开发设置

分叉存储库
创建要素分支： git checkout -b feature/amazing-feature
安装依赖项： pip install -r requirements-dev.txt
进行更改
运行测试： pytest
提交更改： git commit -m 'Add amazing feature'
推送到分支： git push origin feature/amazing-feature
打开拉取请求

📄 许可证

此项目根据MIT许可证获得许可-请参阅许可证文件以获取详细信息。

🆘 支持

文档: https://financenlp-mcp.readthedocs.io
问题:
讨论:
电子邮件: support@financenlp-mcp.com

🏆 致谢

模型上下文协议规范团队
FastAPI框架贡献者
金融NLP研究社区
开源贡献者

______________________________________________________________________

内置于❤️ 金融科技界

Financenlp MCP Server

快速接入

详细介绍

FinanceNLP MCP：用于金融自然语言处理的模型上下文协议服务器

🎯 概述

关键价值主张

🏗️ 建筑

模型上下文协议（MCP）合规性

核心组件

🚀 快速开始

先决条件

安装

Docker部署

📚 API文档

核心MCP端点

POST /mcp/process

工具特定API

1.财务文本摘要

2.财务文本分类

3.金融实体提取

4.金融情绪分析

市场数据集成

GET /market/quote/{symbol}

系统端点

GET /

GET /health

GET /tools

🔧 配置

环境变量

自定义配置

🧪 测试

单元测试

集成测试

手动测试

🔌 集成示例

Python客户端

JavaScript/TypeScript客户端

LangChain集成

📊 性能优化

缓存策略

批处理

异步处理

🔒 安全考虑

API安全

数据隐私

🚀 部署

生产部署

Docker Compose

Kubernetes部署

监测和可观察性

普罗米修斯指标

健康检查

🤝 贡献

开发设置

📄 许可证

🆘 支持

🏆 致谢

目录标签

接入字段

权限和风险

来源信息

继续浏览同类 MCP

`POST /mcp/process`

`GET /market/quote/{symbol}`

`GET /`

`GET /health`

`GET /tools`