亚马逊基岩知识库MCP服务器(非官方)
Amazon Bedrock Knowledge Base管理MCP(Model Context Protocol)服务器。
在本项目中FastMCP使用框架Bedrock Knowledge Base 的操作和 RAG(Retrieval-Augmented Generation)提供功能。

仓库
https://github.com/r3-yamauchi/bedrock-kb-mcp-server
起动方法
uvx --from git+https://github.com/r3-yamauchi/bedrock-kb-mcp-server bedrock-kb-mcp-server主要功能
- Knowledge Base管理:创建、列表检索、详细检索、更新(未实施删除功能)
- 存储类型为S3 仅支持 - 支持自定义分割设置和拆分设置
- 数据源管理:创建、检索列表(未实施删除功能)
- 支持自定义分割设置和拆分设置
- 数据导入作业:启动、状态确认
- RAG查询: Knowledge Base执行搜索查询
- S3文档管理:上载,获取列表
项目结构
bedrock-kb-mcp-server/
├── pyproject.toml # プロジェクト設定と依存関係
├── README.md # プロジェクトドキュメント
├── LICENSE # MITライセンス
├── .gitignore # Git除外設定
└── src/
└── bedrock_kb_mcp_server/
├── __init__.py # パッケージ初期化ファイル
├── main.py # MCPサーバーのメインエントリーポイント
├── bedrock_client.py # AWS Bedrock APIクライアントラッパー
├── models.py # Pydanticモデル(バリデーションと型定義)
├── types.py # TypedDict定義(型ヒントの改善)
└── utils.py # ユーティリティ関数(設定、エラーハンドリング、ログ、ARN正規化)技术栈
核心库
- FastMCP (
>=0.1.0): MCP构建服务器的框架 - 肉毒杆菌3 (
>=1.26.0): AWS SDK for Python - AWS用于与服务通信 - 皮丹提克 (
>=2.0.0):数据验证和设置管理
开发工具
- pytest (
>=7.0):测试框架 - pytest异步 (
>=0.21.0:异步测试支持 - 黑色 (
>=23.0:代码形成器 - 颈毛 (
>=0.1.0):快速Python林特 - 米皮 (
>=1.0):静态类型检查器
Python要件
- Python 3.12以上需要
快速启动
前提条件
- Python 3.12已安装以上内容
uv已安装- AWS已设置帐户和适当的凭据
1.安装相关性
在项目根目录中执行以下命令:
uv sync在开发环境中:
uv sync --all-extras2. AWS设置凭据
通过以下方法之一AWS设置凭据:
方法1:AWS使用配置文件(建议)
export AWS_PROFILE=your-profile-name
export AWS_REGION=us-east-1方法2:环境变数で直接设定
export AWS_ACCESS_KEY_ID=your-access-key-id
export AWS_SECRET_ACCESS_KEY=your-secret-access-key
export AWS_REGION=us-east-1方法3:AWS CLI中所述的工具,调整墙的布局和几何形状
AWS CLI的aws configure中描述的场景,使用以下步骤创建明细表,以便在概念设计中分析体量的体积。
3.设置环境变量(可选)
设置日志级别和结构化日志:
# ログレベルを設定(DEBUG, INFO, WARNING, ERROR, CRITICAL)
export FASTMCP_LOG_LEVEL=INFO
# 構造化ログ(JSON形式)を使用する場合
export FASTMCP_STRUCTURED_LOG=false4.启动服务器
使用以下命令启动服务器:
uv run bedrock-kb-mcp-server成功启动后,将显示以下日志:
2024-01-01 12:00:00,000 - bedrock_kb_mcp_server.main - INFO - Starting Amazon Bedrock Knowledge Base MCP Server5.动作确认
MCP Server从标准输入(stdin)开始JSON-RPC接收格式请求并将其转换为标准输出JSON-RPC返回格式的响应。
行为确认方法1:确认服务器是否启动
启动服务器时,将显示以下日志:。显示此日志后,服务器已成功启动。
动作确认方法2:MCP使用客户端(建议)
MCP支持的客户端(例如:Claude Desktop、Cursor IDE等已弃用的函数的缺少的支持。
Claude Desktop的情况下:
~/Library/Application Support/Claude/claude_desktop_config.json编辑- MCP Server添加:
{
"mcpServers": {
"bedrock-kb-mcp-server": {
"command": "uv",
"args": ["run", "--directory", "/path/to/bedrock-kb-mcp-server", "bedrock-kb-mcp-server"],
"env": {
"AWS_PROFILE": "your-profile-name",
"AWS_REGION": "us-east-1"
}
}
}
}Cursor IDE的情况下:
- 从配置MCP Server添加
- 设置命令路径和环境变量
行为确认方法3:使用测试脚本
包含在项目中test_mcp_server.py使用:
python3 test_mcp_server.py注:此脚本确认服务器将启动并响应请求,但实际AWS API不调用。
安装
uv sync在开发环境中:
uv sync --all-extras使用方法
设置环境变量
必须设置以下环境变量:
AWS_PROFILE: AWS配置文件名(用于管理凭据)AWS_REGION: AWS区域(例如:us-east-1,默认值:us-east-1)FASTMCP_LOG_LEVEL:日志级别(DEBUG,INFO,WARNING,ERROR,CRITICAL,默认值:INFO)FASTMCP_STRUCTURED_LOG:是否使用结构化日志(JSON格式)(true/false,默认值:false)
启动服务器
uv run bedrock-kb-mcp-server主要组件
1. bedrock_client.py - AWS Bedrock API客户机
AWS Bedrock Knowledge Base API中描述的场景,使用以下步骤创建明细表,以便在概念设计中分析体量的体积。
BedrockKBClient 班级
环境变数AWS_REGION从中获取区域AWS初始化客户端:
bedrock-agent: Knowledge Base和数据源管理bedrock-agent-runtime: RAG查询执行s3: S3文档管理
以下设置适用于所有客户端:
- 重试设置(最多3次adaptive模式)
- 连接超时(10秒)
- 读取超时(30秒)
主要方法
Knowledge Base管理
create_knowledge_base():新建Knowledge Base创建list_knowledge_bases():所有Knowledge Base一览获取(支持页面对齐)get_knowledge_base():特定Knowledge Base获取详细信息update_knowledge_base(): Knowledge Base名称、说明、IAM更新角色
数据源管理
create_data_source(): Knowledge Base添加数据源list_data_sources():指定的Knowledge Base获取数据源列表
数据导入作业管理
start_ingestion_job():从数据源Knowledge Base开始数据导入作业get_ingestion_job():获取捕获作业的状态和统计信息
RAG查询
retrieve(): Knowledge Base对于RAG执行查询(可指定结果数1-100)
S3文档管理
upload_document_to_s3():打开本地文件S3上载到时段list_s3_documents(): S3获取时段中的文档列表(可通过前缀过滤)
2. main.py - MCP服务器主机
FastMCP使用框架MCP构建服务器BedrockKBClient功能MCP作为工具公开。
MCP工具
Knowledge Base管理工具
create_knowledge_base: Knowledge Base创建
- 存储类型:S3、S3_VECTORS - 嵌入模型:Amazon Titan、Cohere、Amazon Nova Multimodal Embeddings v1 - 分区设置:BEDROCK_FOUNDATION_MODEL、BEDROCK_DATA_AUTOMATION - 拆分设置:FIXED_SIZE、HIERARCHICAL、SEMANTIC、NONE - 多模式存储配置 - S3 ARN格式和S3 URI同时支持格式 - IAM角色ARN的帐户ID自动完成
list_knowledge_bases:所有Knowledge Base获取列表get_knowledge_base:特定Knowledge Base获得详细信息update_knowledge_base: Knowledge Base更新
数据源管理工具
create_data_source:创建数据源
- 支持分区和拆分设置 - S3 ARN格式和S3 URI同时支持格式
list_data_sources:获取数据源列表
数据导入工具
start_ingestion_job:启动导入作业get_ingestion_job:获取捕获作业的状态
RAG查询工具
retrieve: Knowledge Base对于RAG执行查询
S3文档管理工具
upload_document_to_s3: S3上传文档list_s3_documents: S3获取时段中的文档列表
3. models.py - Pydantic模型
提供请求/响应验证和类型安全性Pydantic定义模型。
StorageType:存储类型枚举类型(S3,S3_VECTORS)SourceType:数据源类型的枚举类型(S3)ParsingStrategy:解析策略枚举类型(BEDROCK_FOUNDATION_MODEL,BEDROCK_DATA_AUTOMATION)ChunkingStrategy:拆分策略枚举类型(FIXED_SIZE,HIERARCHICAL,SEMANTIC,NONE)ParsingConfiguration:分割设置模型ChunkingConfiguration:拆分配置模型VectorIngestionConfiguration:向量导入设置模型CreateKnowledgeBaseRequest: Knowledge Base创建请求验证
- S3 URI形式支持(自动ARN转换为格式) - IAM角色ARN的帐户ID自动完成
CreateDataSourceRequest:验证数据源创建请求
- S3 URI形式支持(自动ARN转换为格式)
- 各种响应模型
4. types.py - TypedDict定义
API提高响应类型安全性TypedDict提供定义。
KnowledgeBaseResponseDict: Knowledge Base创建/更新响应DataSourceResponseDict:数据源创建响应IngestionJobResponseDict:捕获作业响应- 其他响应类型定义
5. utils.py -实用程序函数
配置管理、错误处理、日志输出、ARN提供标准化等通用功能。
validate_aws_credentials(): AWS验证凭据get_log_level():安全获取日志级别handle_errors():错误处理解码器(AWS API错误的正确处理)
- 10种以上AWS支持错误代码 - AWS请求ID的详细错误信息
get_aws_account_id(): STS来修改标记元素的显示属性AWS账户ID获得normalize_s3_arn_or_uri(): S3 URI格式ARN转换为格式normalize_iam_role_arn(): IAM角色ARN的帐户ID自动完成validate_required_string():必需字符串参数验证通用化StructuredFormatter:结构化日志形成器(JSON格式)sanitize_log_data():隐藏敏感信息setup_logging():集中管理记录设置
工作流示例
1. Knowledge Base创建和配置
- S3将文档上载到时段
upload_document_to_s3(local_file_path, bucket_name, s3_key)- Knowledge Base(S3URI格式和IAM角色ARN的缩进格式)
# 基本的なKnowledge Base
create_knowledge_base(
name="My Knowledge Base",
description="Example KB",
role_arn="role/BedrockKBRole", # アカウントIDなし形式(自動補完)
storage_type="S3",
bucket_arn="s3://my-bucket" # S3 URI形式
)
# S3 Vectorsを使用したKnowledge Base
create_knowledge_base(
name="Vector KB",
description="Vector search enabled KB",
role_arn="arn:aws:iam::123456789012:role/BedrockKBRole",
storage_type="S3_VECTORS",
bucket_arn="s3://vector-bucket",
embedding_model_arn="arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v1"
)
# マルチモーダルKnowledge Base(Amazon Nova Multimodal Embeddings v1)
create_knowledge_base(
name="Multimodal KB",
description="KB with Nova Multimodal Embeddings",
role_arn="role/BedrockKBRole",
storage_type="S3_VECTORS",
bucket_arn="s3://vector-bucket",
embedding_model_arn="arn:aws:bedrock:us-east-1::foundation-model/amazon.nova-2-multimodal-embeddings-v1:0",
multimodal_storage_s3_uri="s3://multimodal-storage-bucket/"
)
# カスタムパーシングとチャンキング設定を使用
create_knowledge_base(
name="Custom KB",
description="KB with custom parsing and chunking",
role_arn="role/BedrockKBRole",
storage_type="S3",
bucket_arn="s3://my-bucket",
parsing_strategy="BEDROCK_FOUNDATION_MODEL",
parsing_model_arn="arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-sonnet-20240229-v1:0",
parsing_modality="MULTIMODAL",
chunking_strategy="FIXED_SIZE",
chunking_max_tokens=1000,
chunking_overlap_percentage=20
)- 创建数据源(可使用S3URI格式和自定义设置)
# 基本的なデータソース
create_data_source(
knowledge_base_id="KB123",
name="My Data Source",
source_type="S3",
bucket_arn="s3://my-bucket" # S3 URI形式
)
# カスタムパーシングとチャンキング設定を使用
create_data_source(
knowledge_base_id="KB123",
name="Custom Data Source",
source_type="S3",
bucket_arn="s3://my-bucket",
parsing_strategy="BEDROCK_FOUNDATION_MODEL",
parsing_model_arn="arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-sonnet-20240229-v1:0",
chunking_strategy="FIXED_SIZE",
chunking_max_tokens=1000,
chunking_overlap_percentage=20
)- 启动数据捕获作业
start_ingestion_job(knowledge_base_id, data_source_id)- 检查作业状态
get_ingestion_job(knowledge_base_id, data_source_id, ingestion_job_id)2. RAG执行查询
- Knowledge Base查询
retrieve(knowledge_base_id, query, number_of_results)- 结果包括相关文档和引用信息
体系结构特征
1.分层体系结构
- 显示层:
main.py- MCP定义和发布工具 - 业务逻辑层:
bedrock_client.py- AWS API通讯 - 数据层:
models.py-数据验证和类型定义
2.错误处理
- 所有AWS API呼叫是
ClientError妥善处理 - 10种以上AWS支持错误代码(InternalServerException、InvalidParameterException、ResourceNotFoundException等)
- AWS请求ID的详细错误信息
- 提供符合错误代码的日语消息
- 错误信息被记录在日志中,并向用户返回易懂的消息
3.记录
- 标准日志:人类可读的格式(默认设置)
- 结构化日志: JSON格式输出(
FASTMCP_STRUCTURED_LOG=true启用) - 自动隐藏敏感信息: ARN、认证信息等被自动屏蔽
- 可通过环境变量控制日志级别
4. 型安全性
- Pydantic使用的数据验证
- TypedDict使用API响应类型定义
- 所有MCP工具函数和BedrockKBClient具体定义方法的返回值类型
- 使用类型提示
- mypy支持静态类型检查
- IDE的补充功能和模具安全性提高
5.输入验证和规范化
- ARN格式和URI同时支持格式
- S3 ARN形式: arn:aws:s3:::bucket-name - S3 URI形式: s3://bucket-name 或 s3://bucket-name/path(自动ARN转换为格式) - IAM角色ARN: 完整或缩短(帐户ID自动完成)
- 检查必填字段
- 通过列举存储类型、分区策略和拆分策略来提高类型安全性
- Pydantic模型全面验证
6.重试逻辑
- AWS API在调用中实现重试机制
- 使用自适应重试模式
- 自动重试临时网络错误或速率限制错误
安全注意事项
- AWS认证情报:环境变数
AWS_PROFILE使用管理凭据 - IAM角色: Knowledge Base适宜的IAM需要角色
- S3访问: S3需要对时段的访问权限
- 区域设置:适当的AWS指定区域
- 隐藏机密信息:在日志输出时ARN、认证信息等被自动屏蔽
故障排除
常见问题
- AWS认证错误
- AWS_PROFILE确认是否正确设置 - AWS确认认证信息是否有效 - boto3(IAM角色、EC2实例配置文件),即使没有显式设置,也可以正常工作
- 区域错误
- AWS_REGION确认是否正确设置 - Bedrock确认在该区域是否可用
- 权限错误
- IAM确认角色是否具有所需权限 - S3确认是否具有对时段的访问权限
- 捕获作业失败
- 检查数据源设置 - S3检查时段中的文档格式
- 验证错误
- ARN如果格式正确,请检查错误消息 - 确认是否指定了所有必需字段
错误:AWS找不到凭据
WARNING - AWS認証情報が明示的に設定されていません。解决方法: AWS请设置凭据(请参阅快速启动步骤2)
错误:找不到模块
ModuleNotFoundError: No module named 'fastmcp'解决方法:请安装依赖关系
uv sync错误:Python版本旧
ERROR: This package requires Python >=3.12解决方法: Python 3.12请安装以上内容
服务器未启动
- 日志级别
DEBUG中描述的场景,使用以下步骤创建明细表,以便在概念设计中分析体量的体积
export FASTMCP_LOG_LEVEL=DEBUG
uv run bedrock-kb-mcp-server- 启用结构化日志JSON形式确认:
export FASTMCP_STRUCTURED_LOG=true
uv run bedrock-kb-mcp-server实际使用示例
Knowledge Base获取列表
MCP从客户端list_knowledge_bases如果您想要发送文档作为只读文件,请选择AWS帐户中的所有Knowledge Base列表框中,此格式对应于条目“无”。
Knowledge Base创建
{
"name": "my-knowledge-base",
"description": "テスト用のKnowledge Base",
"role_arn": "role/BedrockKnowledgeBaseRole",
"storage_type": "S3",
"bucket_arn": "s3://my-documents-bucket"
}注意:
role_arn是缩短形式(role/ROLE_NAME)也可以使用,帐户ID自动完成bucket_arnIcult支持的pambans;3种巴比伦人s3://bucket-name)也可以使用,自动ARN转换为格式
RAG执行查询
{
"knowledge_base_id": "YOUR_KB_ID",
"query": "ドキュメントの内容について教えてください",
"number_of_results": 5
}注意事项
- 实际的AWS API中所述修改相应参数的值AWS认证信息和IAM需要权限
- Knowledge Base创建适当的IAM需要角色
- S3需要对时段的访问权限
- 根据区域的不同Bedrock时褪色为此颜色
开発
代码格式
black src/
ruff check src/
mypy src/测试(今后计划实施)
pytest许可证
MIT许可证-请参阅 许可证 文件以获取详细信息。
