知识库MCP服务器

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

功能

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

快速开始

1. 从控制台获取您的AppBuilder API密钥。
2. 将授权令牌格式化为：Bearer+<AppBuilder API密钥>（在中间保留"+"）。
3. 配置如下：

{
  "mcpServers": {
    "knowledge_base": {
      "url": "http://appbuilder.baidu.com/v2/knowledgeBase/mcp/sse?api_key=Bearer+bce-v3/ALTAK..."
    }
  }
}

可用工具

create_knowledge_base

创建一个新的知识库。

参数：

• name (str)：知识库名称。
• description (str)：知识库描述。

返回值：

• 包含知识库ID、名称、描述和请求ID的字典。

describe_knowledge_base

获取特定知识库的详细信息。

参数：

• id (str)：知识库ID。

返回值：

• 包含知识库ID、名称和描述的字典。

list_knowledge_bases

列出所有知识库。

参数：

• max_keys (int, 可选)：要返回的知识库的最大数量，默认为110。

返回值：

• 字典列表，每个字典包含一个知识库的ID、名称和描述。

query_knowledge_base

从知识库中查询内容。

参数：

• query (str)：搜索查询字符串。
• id_list (list[str])：要在其中搜索的知识库ID列表。

返回值：

• 字典列表，每个字典包含一个片段的ID、内容、文档ID和文档名称。

upload_document

将文档上传到知识库。

参数：

• id (str)：知识库ID。
• file_path (str)：要上传的文件路径。
• template (str, 可选)：可选处理模板。选项："ppt"、"paper"、"resume"或"default"。默认为"default"。

返回值：

• 包含文档ID和请求ID的字典。

list_documents

获取知识库中的所有文档。

参数：

• 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)

    # 创建知识库案例
    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

    # 描述知识库案例
    appbuilder.logger.debug(f"创建知识库成功: {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"描述知识库成功: {knowledge_base_id}")

    # 列出知识库案例
    result = await mcp_client.call_tool(
        "list_knowledge_bases",
        {"max_keys": 10},
    )
    assert len(result.content) == 10
    appbuilder.logger.debug(f"列出知识库成功: {len(result.content)}")

    # 上传文档案例
    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"上传文档成功: {document_id}")

    # 列出文档案例
    result = await mcp_client.call_tool(
        "list_documents",
        {"id": knowledge_base_id},
    )
    print(result)
    appbuilder.logger.debug(f"列出文档成功: {knowledge_base_id}")

    # 查询知识库案例
    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("查询知识库成功")


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())