AI导哈

知识库管理服务器

Knowledge Base MCP Server 提供了一套通过MCP协议管理知识库和文档的工具，支持创建、查询和管理知识库及其内容。

  • by baidubce
  • Python
  • 云端部署
  • 587 星标
  • 更新于
在 GitHub 查看 baidubce/app-builder

使用说明

知识库MCP服务器

知识库MCP服务器提供了一套通过MCP协议管理知识库和文档的工具。它允许您创建、查询和管理知识库及其内容。

特性

  • 知识库创建与管理：创建新的知识库并列出所有可用的知识库。
  • 文档管理：将文件上传到知识库，并列出知识库中的所有文档。
  • 知识库查询：使用查询字符串搜索知识库。

快速开始

  1. 从控制台获取AppBuilder API密钥
  2. 将您的授权令牌格式化为： Bearer+<AppBuilder API Key> （保持“+”在中间）
  3. 配置为
{
  "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())

相关 MCP