知识库MCP服务器
知识库MCP服务器提供了一套通过MCP协议管理知识库和文档的工具。它允许您创建、查询和管理知识库及其内容。
特性
- 知识库创建与管理：创建新的知识库并列出所有可用的知识库。
- 文档管理：将文件上传到知识库，并列出知识库中的所有文档。
- 知识库查询：使用查询字符串搜索知识库。
快速开始
- 从控制台获取AppBuilder API密钥
- 将您的授权令牌格式化为：
Bearer+（保持“+”在中间） - 配置为
{
"mcpServers": {
"knowledge_base": {
"url": "http://appbuilder.baidu.com/v2/knowledgeBase/mcp/sse?api_key=Bearer+bce-v3/ALTAK..."
}
}
}可用工具
创建知识库
创建新的知识库。
参数：
name（str）：知识库的名称。description（str）：知识库的描述。
退货：
- 包含知识库ID、名称、描述和请求ID的词典。
描述知识库
获取特定知识库的详细信息。
参数：
id（str）：知识库的ID。
退货：
- 包含知识库ID、名称和描述的词典。
list_知识库
列出所有知识库。
参数：
max_keys（int，可选）：要返回的知识库的最大数量。默认值为10。
退货：
- 字典列表，每个字典都包含知识库的ID、名称和描述。
查询知识库
从知识库中查询内容。
参数：
query（str）：搜索查询字符串。id_list（list\[str\]）：要搜索的知识库ID列表。
退货：
- 字典列表，每个字典包含一个块的ID、内容、文档ID和文档名称。
upload_document
将文档上传到知识库。
参数：
id（str）：知识库ID。file_path（str）：要上传的文件的路径。template（str，可选）：要使用的可选处理模板。选项：“ppt”、“纸质”、“简历”或“默认”。默认值为“默认”。
退货：
- 包含文档ID和请求ID的字典。
list_文档
获取知识库中的所有文档。
参数：
id（str）：知识库ID。
退货：
- 字典列表，每个字典包含一个文档的ID、名称和字数。
使用示例
import asyncio
import json
import io
import os
import appbuilder
from appbuilder.mcp_server.client import MCPClient
os.environ["APPBUILDER_TOKEN"] = "YOUR_APPBUILDER_TOKEN"
async def main():
mcp_client = MCPClient()
await mcp_client.connect_to_server(service_url=service_url)
# case create_knowledge_base
result = await mcp_client.call_tool(
"create_knowledge_base",
{"name": "mcp测试可删", "description": "mcp测试，可删"},
)
knowledge_base_info = json.loads(result.content[0].text)
knowledge_base_id = knowledge_base_info.get("id")
assert knowledge_base_id is not None
# case describe_knowledge_base
appbuilder.logger.debug(f"create knowledge base success: {knowledge_base_id}")
result = await mcp_client.call_tool(
"describe_knowledge_base",
{"id": knowledge_base_id},
)
knowledge_base_info = json.loads(result.content[0].text)
knowledge_base_id = knowledge_base_info.get("id")
assert knowledge_base_id is not None
appbuilder.logger.debug(f"describe knowledge base success: {knowledge_base_id}")
# case list_knowledge_bases
result = await mcp_client.call_tool(
"list_knowledge_bases",
{"max_keys": 10},
)
assert len(result.content) == 10
appbuilder.logger.debug(f"list knowledge bases success: {len(result.content)}")
# case upload_document
file_content = "这里是你的文件内容字符串"
with io.BytesIO(file_content.encode("utf-8")) as f:
result = await mcp_client.call_tool(
"upload_document",
{"id": knowledge_base_id, "file_data": f, "file_name": "test.txt"},
)
document_info = json.loads(result.content[0].text)
document_id = document_info.get("documentId")
assert document_id is not None
appbuilder.logger.debug(f"upload document success: {document_id}")
# case list_documents
result = await mcp_client.call_tool(
"list_documents",
{"id": knowledge_base_id},
)
print(result)
appbuilder.logger.debug(f"list documents base success: {knowledge_base_id}")
# case query_knowledge_base
result = await mcp_client.call_tool(
"query_knowledge_base",
{"query": "分子", "id_list": [knowledge_base_id]},
)
assert result.content[0].text is not None
appbuilder.logger.debug("query knowledge base success")
if __name__ == "__main__":
service_url = (
"http://appbuilder.baidu.com/v2/ai_search/mcp/sse?api_key="
+ os.environ.get("APPBUILDER_TOKEN")
)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())