多协议服务器

提供.NET和Python实现的多协议服务器,支持Cosmos、Blob和SQL数据源的查询与管理。

  • by csdmichael
  • C#
  • 本地部署
  • 更新于
在 GitHub 查看 csdmichael/SalesPOC.MCP

使用说明

MCP服务器:。NET和Python

主要项目

  • 销售POC用户界面:https://github.com/csdmichael/SalesPOC.UI

此工作区包含用于相同数据源的两个MCP服务器实现:

  1. .NET MCP服务器 (ASP.NET核心+MCP C#SDK)
  2. Python MCP服务器

代理快速入门(工具布线)

当代理需要回答用户问题时,请使用此决策指南:

  • 产品搜索/目录数据中的详细信息→ 宇宙 (find_products / get_product_by_id)
  • 产品文档/规格文件→ 二进制大对象 (list_documents / read_document)
  • SQL/表探索或自定义只读报告→ 结构化查询语言 (list_tables / query_data)

如果可能,从狭义查询开始(limit过滤器, TOP)并且仅在需要时进行扩展。

.NET MCP服务器

这个。NET实现正在进行中 dotnet/ 并公开HTTP MCP端点。

服务器

  • dotnet/CosmosMcpServer
    • 工具:
      • FindProducts(searchText?, category?, minPrice?, maxPrice?, limit?)
      • GetProductById(id)
      • DownloadProducts(searchText?, category?, minPrice?, maxPrice?, limit?, maxChars?)
  • dotnet/BlobMcpServer
    • 工具:
      • ListDocuments(prefix?, limit?)
      • ReadDocument(blobName, maxChars?)
      • DownloadDocument(blobName, maxChars?)
  • dotnet/SqlMcpServer
    • 工具:
      • ListTables(schema?, limit?)
      • ListTableColumns(table, schema?)
      • QueryTableRows(table, filters?, schema?, orderBy?, descending?, maxRows?)
      • QueryData(sql, maxRows?) (只读 SELECT/WITH 仅)

.NET工具参考(适用于代理)

Cosmos MCP(dotnet/CosmosMcpServer)

  • FindProducts(searchText?, category?, minPrice?, maxPrice?, limit?)
    • 用于:产品发现和筛选搜索
    • 输入:
      • searchText:匹配产品名称/描述(包含不区分大小写的内容)
      • category:精确类别筛选器
      • minPrice / maxPrice:数字价格界限
      • limit:夹紧 1..100
    • 输出形状: { count, items[] } (项目字段包括 id, name, category, price, description)
  • GetProductById(id)
    • 用于:按id精确查找产品
    • 输入: id 需要,修剪
    • 输出形状:完整的产品对象或 null
  • DownloadProducts(searchText?, category?, minPrice?, maxPrice?, limit?, maxChars?)
    • 用于:将产品搜索结果导出为可下载的JSON文档有效负载
    • 输入:
      • 与相同的过滤器 FindProducts
      • maxChars:max返回base64字符(有效载荷是base64编码的JSON)
    • 输出形状: { file_name, content_type, encoding, truncated, content, size_bytes, count }

Blob MCP(dotnet/BlobMcpServer)

  • ListDocuments(prefix?, limit?)
    • 用于:读取前发现可用文件
    • 输入:
      • prefix:可选blob路径/前缀过滤器
      • limit:夹紧 1..200
    • 输出形状: { count, items[] } (项目字段包括 name, size, content_type, last_modified)
  • ReadDocument(blobName, maxChars?)
    • 用于:检索一个blob内容
    • 输入:
      • blobName:需要精确的blob路径/名称
      • maxChars:夹紧 1..200000
    • 输出形状: { blob_name, content_type, encoding, truncated, content, size_bytes }
    • 行为:
      • 可解码时返回UTF-8文本
      • 否则返回Base64
  • DownloadDocument(blobName, maxChars?)
    • 用于:下载完整blob作为base64有效载荷(适用于二进制文档)
    • 输入:
      • blobName:需要精确的blob路径/名称
      • maxChars:max返回base64字符
    • 输出形状: { blob_name, file_name, content_type, encoding, truncated, content, size_bytes }

SQL-MCP(dotnet/SqlMcpServer)

  • ListTables(schema?, limit?)
    • 用于:查询构建前的架构发现
    • 输入:
      • schema:默认值 dbo
      • limit:夹紧 1..1000
    • 输出形状: { count, items[] } 每个项目在哪里 { schema, table }
  • ListTableColumns(table, schema?)
    • 用于:发现有效列以构建安全的筛选查询
    • 输入:
      • table:必需的表名
      • schema:默认值 dbo
    • 输出形状: { schema, table, count, columns[] }
  • QueryTableRows(table, filters?, schema?, orderBy?, descending?, maxRows?)
    • 用于:使用相等筛选器从任何表中检索代理安全的行
    • 输入:
      • table:必需的表名
      • filters:可选对象 { columnName: value } 相等过滤器
      • schema:默认值 dbo
      • orderBy:可选排序列
      • descending:排序方向切换
      • maxRows:夹紧 1..2000
    • 输出形状: { schema, table, count, columns[], truncated, items[] }
    • 安全约束:
      • 架构/表/列必须是有效的标识符
      • 过滤器和 orderBy 必须引用目标表中的实际列
  • QueryData(sql, maxRows?)
    • 用于:只读分析/数据检索查询
    • 输入:
      • sql: 单个 仅声明;必须从以下内容开始 SELECT 或 WITH
      • maxRows:夹紧 1..2000
    • 输出形状: { count, columns[], truncated, items[] }
    • 安全约束:
      • 阻止写入/DDL/DCL关键字(insert, update, delete, drop, alter等等)
      • 拒绝多语句SQL

认证

全部。NET服务器使用以下方式对Azure服务进行身份验证 DefaultAzureCredential (托管身份、Azure CLI、环境凭据等)。带有嵌入式密钥或密码的连接字符串是 不 支持。对于SQL,连接字符串必须使用 Authentication=Active Directory Default 并且必须 不 包括用户名或密码。

所需的环境变量

  • COSMOS_ENDPOINT --Cosmos DB帐户端点URL(例如。 https://<account>.privatelink.documents.azure.com:443/)
  • COSMOS_DATABASE_NAME (可选,默认值: sales)
  • COSMOS_CONTAINER_NAME (可选,默认值: products)
  • AZURE_BLOB_ACCOUNT_URL --Blob存储帐户URL(例如。 https://<account>.privatelink.blob.core.windows.net)
  • AZURE_BLOB_CONTAINER_NAME (可选,默认值: semiconductor-product-documents)
  • SQL_CONNECTION_STRING --必须使用 Authentication=Active Directory Default;不允许使用用户名/密码

运行(.NET)

dotnet run --project .\dotnet\CosmosMcpServer\CosmosMcpServer.csproj
dotnet run --project .\dotnet\BlobMcpServer\BlobMcpServer.csproj
dotnet run --project .\dotnet\SqlMcpServer\SqlMcpServer.csproj

默认本地MCP端点:

  • 宇宙: http://localhost:6204
  • Blob: http://localhost:6128
  • SQL: http://localhost:6275

的路由前缀。NET服务器: / (根路径)。

测试URL(.NET MCP)

在MCP客户端(HTTP传输)中使用这些URL来测试部署的服务器:

  • Blob MCP: https://mcp-salespoc-blob.azurewebsites.net
  • Cosmos MCP: https://mcp-salespoc-cosmos.azurewebsites.net
  • SQL-MCP: https://mcp-salespoc-sql.azurewebsites.net

如果您的MCP客户端需要路由路径,请附加配置的MCP路径(例如 /mcp 如果在客户端/服务器设置中配置)。


Python MCP服务器

Python实现在 src/.

服务器

  • src/cosmos_server.py
    • find_products:按文本/类别和可选价格范围搜索
    • get_product_by_id:按以下方式获取一个产品 id
  • src/blob_server.py
    • list_documents:按可选前缀列出blob文档
    • read_document:将blob读取为文本(二进制为base64)
  • src/sql_server.py
    • list_tables:列出架构中的表
    • query_data:以只读方式运行 SELECT/WITH 查询

Python工具参考(用于代理)

Python工具在语义上与对齐。NET,使用snake_case名称:

  • 宇宙(src/cosmos_server.py)
    • find_products(search_text?, category?, min_price?, max_price?, limit=20)
    • get_product_by_id(id)
  • Blob(src/blob_server.py)
    • list_documents(prefix?, limit=50)
    • read_document(blob_name, max_chars=20000)
  • SQL(src/sql_server.py)
    • list_tables(schema='dbo', limit=200)
    • query_data(sql, max_rows=200)

Python特定的行为对代理很重要:

  • 运输可以作为 stdio, sse,或 streamable-http 通过 MCP_TRANSPORT
  • Cosmos和Blob使用以下方式进行身份验证 DefaultAzureCredential (托管身份、Azure CLI等)
  • SQL使用 Active Directory Default 通过验证 SQL_CONNECTION_STRING;用户名/密码被拒绝
  • 共享保护 src/policies.py 强制执行:
    • 速率限制
    • 提示内容检查
    • 输出尺寸整形

代理提示示例→ 推荐工具

  • “查找价格低于400美元的传感器芯片”→ 宇宙 find_products / FindProducts
  • “显示Chip-101的文档”→ Blob list_documents 然后 read_document
  • “dbo中有哪些表?”→ SQL list_tables / ListTables
  • “从SQL运行快速产品报告”→ SQL query_data / QueryData

故障排除(常见)

  • tools/list 工作,但 tools/call 失败:
    • 通常缺少连接环境变量
  • 超文本传输协议 503 在已部署的应用程序上:
    • 应用程序正在启动/重新启动或崩溃;检查应用服务日志流
  • SQL查询被阻止:
    • 查询违反了只读策略(非SELECT、被阻止的关键字或多条语句)

共享运行时策略(Python服务器)

Python服务器从以下位置执行共享运行时策略 src/policies.py:

  • 通过以下方式控制输出尺寸 POLICY_MAX_OUTPUT_CHARS (默认值 12000)
  • 快速注射模式堵塞
  • 有害/仇恨内容模式屏蔽
  • 每个工具在内存中的速率限制 POLICY_RATE_LIMIT_PER_MINUTE (默认值 60)

策略环境变量:

  • POLICY_MAX_OUTPUT_CHARS=12000
  • POLICY_RATE_LIMIT_PER_MINUTE=60

设置(Python)

  1. 创建并激活虚拟环境:
    python -m venv .venv
    .\.venv\Scripts\Activate.ps1
    
  2. 安装依赖项:
    pip install -r requirements.txt
    
  3. 设置环境变量(PowerShell):
    $env:COSMOS_ENDPOINT="https://<your-account>.privatelink.documents.azure.com:443/"
    $env:COSMOS_DATABASE_NAME="sales"
    $env:COSMOS_CONTAINER_NAME="products"
    $env:AZURE_BLOB_ACCOUNT_URL="https://<your-account>.privatelink.blob.core.windows.net"
    $env:AZURE_BLOB_CONTAINER_NAME="semiconductor-product-documents"
    $env:SQL_CONNECTION_STRING="Driver={ODBC Driver 18 for SQL Server};Server=tcp:<server>.privatelink.database.windows.net,1433;Database=<database>;Authentication=Active Directory Default;Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30;"
    $env:POLICY_MAX_OUTPUT_CHARS="12000"
    $env:POLICY_RATE_LIMIT_PER_MINUTE="60"
    

运行(Python)

python src/cosmos_server.py
python src/blob_server.py
python src/sql_server.py

许可证

该项目根据 MIT许可证.