ZeroDB MCP Python客户端SDK

](https://badge.fury.io/py/zerodb-mcp) ![Python Support](https://pypi.org/project/zerodb-mcp/) ![License: MIT](https://opensource.org/licenses/MIT)

用于ZeroDB MCP Bridge API的可用于生产的Python客户端。提供对所有60多种操作的全面异步访问，包括向量、TurboQuant向量压缩（PolarQuant+QJL）、NoSQL表、文件存储、事件、RLHF和管理功能。

特性

API全面覆盖：所有60+MCP网桥操作
异步/等待支持：基于 httpx 用于高性能异步操作
类型安全：完整的Pydantic模型验证
自动检索：具有指数回退的智能重试逻辑
错误处理：全面的异常层次结构
速率限制：内置速率限制处理，支持后重试
认证：API密钥和JWT令牌支持
生产就绪：经过战斗测试，测试覆盖率超过90%

安装

pip install zerodb-mcp

开发安装

git clone https://github.com/ainative/zerodb-mcp-python.git
cd zerodb-mcp-python
pip install -e ".[dev]"

快速开始

使用ZeroDB云

import asyncio
from zerodb_mcp import ZeroDBClient

async def main():
    # Initialize client (cloud)
    client = ZeroDBClient(api_key="your_api_key")

    # Create a project
    project = await client.projects.create(
        name="My AI Project",
        tier="pro"
    )

    # Upsert vectors
    result = await client.vectors.upsert(
        project_id=project["project_id"],
        embedding=[0.1, 0.2, 0.3] * 512,  # 1536 dimensions
        document="Machine learning tutorial on neural networks",
        metadata={"category": "education", "language": "en"}
    )

    # Search vectors
    results = await client.vectors.search(
        project_id=project["project_id"],
        query_vector=[0.15, 0.25, 0.35] * 512,
        limit=10,
        threshold=0.7
    )

    for item in results["results"]:
        print(f"Document: {item['document']}")
        print(f"Similarity: {item['similarity']:.3f}\n")

    # Close client
    await client.close()

# Run async code
asyncio.run(main())

使用ZeroDB本地

import asyncio
from zerodb_mcp import ZeroDBClient

async def main():
    # Initialize client (local)
    client = ZeroDBClient(
        base_url="http://localhost:8000",  # Local ZeroDB instance
        api_key="local-dev-key"  # Any value works for local
    )

    # Same API as cloud - use all operations identically
    project = await client.projects.create(
        name="Local Development Project"
    )

    # Develop offline, sync when ready
    result = await client.vectors.upsert(
        project_id=project["project_id"],
        embedding=[0.1, 0.2, 0.3] * 512,
        document="Test document for local development"
    )

    await client.close()

asyncio.run(main())

了解更多： 看 ZeroDB本地快速入门用于安装和设置。

认证

API密钥验证

# Method 1: Direct initialization
client = ZeroDBClient(api_key="your_api_key")

# Method 2: Environment variable
# Set ZERODB_API_KEY in .env file
from dotenv import load_dotenv
load_dotenv()

client = ZeroDBClient()  # Automatically loads from env

JWT令牌身份验证

# Method 1: Direct initialization
client = ZeroDBClient(jwt_token="your_jwt_token")

# Method 2: Environment variable
# Set ZERODB_JWT_TOKEN in .env file
client = ZeroDBClient()  # Automatically loads from env

核心业务

矢量运算（10次运算）

# 1. Upsert vector
result = await client.vectors.upsert(
    project_id="550e8400-e29b-41d4-a716-446655440000",
    embedding=[0.1, 0.2, ...],
    document="Text content",
    namespace="default",
    metadata={"key": "value"}
)

# 2. Batch upsert
vectors = [
    {"embedding": [0.1, ...], "document": "Doc 1", "metadata": {}},
    {"embedding": [0.2, ...], "document": "Doc 2", "metadata": {}}
]
result = await client.vectors.batch_upsert(project_id, vectors)

# 3. Search vectors
results = await client.vectors.search(
    project_id=project_id,
    query_vector=[0.15, 0.25, ...],
    limit=20,
    threshold=0.8
)

# 4. Delete vector
await client.vectors.delete(project_id, vector_id)

# 5. Get vector
vector = await client.vectors.get(project_id, vector_id)

# 6. List vectors
vectors = await client.vectors.list(project_id, limit=100, offset=0)

# 7. Vector statistics
stats = await client.vectors.stats(project_id)

# 8. Create index
await client.vectors.create_index(project_id, index_type="hnsw")

# 9. Optimize storage
result = await client.vectors.optimize_storage(project_id)

# 10. Export vectors
export = await client.vectors.export(project_id, format="json")

量子运算（6次运算）

ICLR 2026的TurboQuant矢量压缩（PolarQuant+QJL）。在0.9999余弦相似度下实现约3.5倍的压缩。

# 1. Compress vector using TurboQuant (PolarQuant + QJL)
result = await client.quantum.compress_vector(
    project_id=project_id,
    vector_id=vector_id,
    target_dimensions=128,
    backend="ionq"  # or "simulator", "braket"
)

# 2. Decompress vector
result = await client.quantum.decompress_vector(
    project_id=project_id,
    compressed_state=state,
    original_dimensions=1536
)

# 3. TurboQuant hybrid similarity search
results = await client.quantum.hybrid_similarity(
    project_id=project_id,
    query_vector=[0.1, 0.2, ...],
    top_k=10,
    quantum_weight=0.6,
    backend="ionq"
)

# 4. TurboQuant space optimization
result = await client.quantum.optimize_space(
    project_id=project_id,
    target_compression=0.5
)

# 5. TurboQuant feature mapping
result = await client.quantum.feature_map(
    project_id=project_id,
    vector_id=vector_id,
    feature_map_type="ZZFeatureMap"
)

# 6. TurboQuant kernel similarity
result = await client.quantum.kernel_similarity(
    project_id=project_id,
    vector_id_a=vec_a,
    vector_id_b=vec_b,
    kernel_type="fidelity"
)

表操作（8次操作）

# 1. Create table
table = await client.tables.create(
    project_id=project_id,
    table_name="users",
    schema_definition={
        "user_id": "string",
        "email": "string",
        "age": "integer"
    },
    indexes=["user_id", "email"]
)

# 2. Insert rows
result = await client.tables.insert_rows(
    project_id=project_id,
    table_name="users",
    rows=[
        {"user_id": "u1", "email": "user1@example.com", "age": 25},
        {"user_id": "u2", "email": "user2@example.com", "age": 30}
    ]
)

# 3. Query rows
results = await client.tables.query_rows(
    project_id=project_id,
    table_name="users",
    filters={"age": {"$gte": 18}},
    sort_by="age",
    order="desc"
)

# 4. Update rows
result = await client.tables.update_rows(
    project_id=project_id,
    table_name="users",
    filters={"age": {"$gte": 18}},
    updates={"verified": True}
)

# 5. Delete rows
result = await client.tables.delete_rows(
    project_id=project_id,
    table_name="users",
    filters={"active": False}
)

# 6. Get table
table = await client.tables.get(project_id, "users")

# 7. List tables
tables = await client.tables.list(project_id)

# 8. Delete table
await client.tables.delete(project_id, "users")

文件操作（6个操作）

# 1. Upload file from path
result = await client.files.upload(
    project_id=project_id,
    file_path="/path/to/document.pdf",
    metadata={"category": "legal"}
)

# 2. Upload file from bytes
content = b"Hello, World!"
result = await client.files.upload_bytes(
    project_id=project_id,
    content=content,
    file_name="hello.txt",
    content_type="text/plain"
)

# 3. Download file
content = await client.files.download(project_id, file_id)

# Save to file
await client.files.download(project_id, file_id, save_path="/path/to/save.pdf")

# 4. List files
files = await client.files.list(project_id, content_type="application/pdf")

# 5. Get file metadata
metadata = await client.files.get_metadata(project_id, file_id)

# 6. Generate presigned URL
url = await client.files.generate_presigned_url(
    project_id=project_id,
    file_id=file_id,
    expires_in=3600  # 1 hour
)

项目运营（7项运营）

# 1. Create project
project = await client.projects.create(
    name="My Project",
    tier="pro",
    settings={"enable_quantum": True}
)

# 2. Get project
project = await client.projects.get(project_id)

# 3. List projects
projects = await client.projects.list(tier="pro")

# 4. Update project
await client.projects.update(
    project_id=project_id,
    name="Updated Name"
)

# 5. Delete project
await client.projects.delete(project_id, confirm=True)

# 6. Get project stats
stats = await client.projects.get_stats(project_id)

# 7. Enable database
await client.projects.enable_database(
    project_id=project_id,
    database_type="postgres"
)

事件操作（5次操作）

# 1. Create event
event = await client.events.create(
    project_id=project_id,
    event_type="user.login",
    event_data={"user_id": "u123", "ip": "192.168.1.1"}
)

# 2. List events
events = await client.events.list(
    project_id=project_id,
    event_type="user.login",
    start_date="2025-01-01T00:00:00Z"
)

# 3. Get event
event = await client.events.get(project_id, event_id)

# 4. Subscribe to events
subscription = await client.events.subscribe(
    project_id=project_id,
    event_types=["user.login", "user.logout"]
)

# 5. Event statistics
stats = await client.events.stats(project_id)

RLHF作战（10次作战）

# 1. Collect interaction
interaction = await client.rlhf.collect_interaction(
    session_id="sess_123",
    project_id=project_id,
    interaction_type="query",
    user_input="What is AI?",
    agent_response="AI is...",
    context={"model": "gpt-4"}
)

# 2. Collect feedback
feedback = await client.rlhf.collect_agent_feedback(
    session_id="sess_123",
    project_id=project_id,
    interaction_id=interaction["interaction_id"],
    rating=5,
    feedback_text="Very helpful!"
)

# 3. Collect workflow feedback
await client.rlhf.collect_workflow_feedback(
    session_id="sess_123",
    project_id=project_id,
    workflow_id="search",
    success=True,
    duration_ms=250
)

# 4. Report error
await client.rlhf.collect_error_report(
    session_id="sess_123",
    project_id=project_id,
    error_type="ValidationError",
    error_message="Invalid input"
)

# 5-10. Status, summary, session management
status = await client.rlhf.get_status(project_id)
summary = await client.rlhf.get_summary(project_id)
session = await client.rlhf.start_collection(project_id, "sess_123")
await client.rlhf.stop_collection(project_id, "sess_123")
interactions = await client.rlhf.get_session_interactions(project_id, "sess_123")
await client.rlhf.broadcast_event(project_id, "feedback.received", {})

管理操作（5个操作）

需要管理员权限。

# 1. System statistics
stats = await client.admin.get_system_stats()

# 2. List all projects
projects = await client.admin.list_all_projects(tier="enterprise")

# 3. User usage
usage = await client.admin.get_user_usage("user_id")

# 4. System health
health = await client.admin.system_health()

# 5. Optimize database
result = await client.admin.optimize_database(vacuum=True, reindex=True)

错误处理

from zerodb_mcp import (
    ZeroDBError,
    AuthenticationError,
    RateLimitError,
    ValidationError,
    ResourceNotFoundError,
    QuotaExceededError
)

try:
    result = await client.vectors.upsert(...)
except AuthenticationError as e:
    print(f"Authentication failed: {e}")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except ValidationError as e:
    print(f"Validation error: {e.errors}")
except ResourceNotFoundError as e:
    print(f"Not found: {e.resource_type} {e.resource_id}")
except QuotaExceededError as e:
    print(f"Quota exceeded: {e.quota_type} {e.current}/{e.limit}")
except ZeroDBError as e:
    print(f"API error: {e}")

上下文管理器使用情况

async with ZeroDBClient(api_key="your_key") as client:
    project = await client.projects.create(name="Test Project")
    # Client automatically closes when exiting context

配置

环境变量

创建一个 .env 文件：

ZERODB_API_KEY=your_api_key_here
# OR
ZERODB_JWT_TOKEN=your_jwt_token_here

自定义基本URL

client = ZeroDBClient(
    api_key="your_key",
    base_url="https://custom-api.example.com"
)

超时配置

client = ZeroDBClient(
    api_key="your_key",
    timeout=60.0,  # seconds
    max_retries=5,
    retry_delay=2.0
)

高级用法

并行操作

import asyncio

# Execute multiple operations in parallel
results = await asyncio.gather(
    client.vectors.search(project_id, query1),
    client.vectors.search(project_id, query2),
    client.vectors.search(project_id, query3)
)

批处理

# Process large datasets in chunks
from zerodb_mcp.utils import chunk_list

documents = [...]  # Large list
embedding_function = ...  # Your embedding function

for chunk in chunk_list(documents, chunk_size=100):
    vectors = [
        {
            "embedding": embedding_function(doc),
            "document": doc,
            "metadata": {}
        }
        for doc in chunk
    ]

    await client.vectors.batch_upsert(project_id, vectors)

测试

# Run tests
pytest

# Run tests with coverage
pytest --cov=zerodb_mcp --cov-report=html

# Run specific test file
pytest tests/test_vectors.py

发展

# Install dev dependencies
pip install -e ".[dev]"

# Format code
black zerodb_mcp/
isort zerodb_mcp/

# Type checking
mypy zerodb_mcp/

# Linting
flake8 zerodb_mcp/

贡献

分叉存储库
创建要素分支(git checkout -b feature/amazing-feature)
提交您的更改(git commit -m 'Add amazing feature')
推到分支(git push origin feature/amazing-feature)
打开拉取请求

许可证

MIT许可证-请参阅许可证文件以获取详细信息。

ZeroDB集成：免费嵌入+完整RAG

🎉 新增：与LangChain兼容的ZeroDB嵌入

保存 每月100美元以上 免费嵌入HuggingFace，同时获得与OpenAI/Cohere相同的质量！

from zerodb_mcp import ZeroDBClient, ZeroDBEmbeddings

# Initialize FREE embeddings
embeddings = ZeroDBEmbeddings(
    api_key="your-api-key",
    model="BAAI/bge-small-en-v1.5"  # 384 dims, very fast
)

# Embed documents (FREE!)
docs = ["Machine learning tutorial", "Python guide", "Database optimization"]
doc_embeddings = embeddings.embed_documents(docs)

print(f"Generated {len(doc_embeddings)} embeddings")
print(f"Cost: $0.00 (FREE!)")

# Embed query
query = "How to optimize databases?"
query_emb = embeddings.embed_query(query)

# Search vectors
results = await client.vectors.search(
    project_id=project_id,
    query_vector=query_emb,
    limit=5
)

成本比较

提供者	嵌入成本	矢量存储	每月总计
ZeroDB	0.00美元（免费）	$0-29	$0-29
OpenAI+松果	100美元	70美元	170美元
Cohere+Qdrant	100美元	50美元	150美元

每月储蓄：120-150美元 使用ZeroDB！

可用模型

# Fast & Small (384 dimensions, DEFAULT)
embeddings = ZeroDBEmbeddings(model="BAAI/bge-small-en-v1.5")

# Fast & Balanced (384 dimensions)
embeddings = ZeroDBEmbeddings(model="sentence-transformers/all-MiniLM-L6-v2")

# Higher Quality (768 dimensions)
embeddings = ZeroDBEmbeddings(model="sentence-transformers/all-mpnet-base-v2")

完整的RAG示例（5分钟）

import asyncio
from zerodb_mcp import ZeroDBClient, ZeroDBEmbeddings

async def rag_example():
    # Initialize
    client = ZeroDBClient(api_key="your-key")
    embeddings = ZeroDBEmbeddings(api_key="your-key")

    # Create project
    project = await client.projects.create(
        project_name="rag_demo",
        tier="free"  # FREE tier for <100K vectors
    )
    project_id = project["project_id"]

    # Knowledge base
    knowledge = [
        "Python is a versatile programming language used for web, AI, and data science",
        "Machine learning is a subset of AI that learns from data",
        "Neural networks are inspired by biological neurons"
    ]

    # Generate embeddings (FREE!)
    knowledge_embeddings = embeddings.embed_documents(knowledge)

    # Store vectors
    vectors = [
        {
            "embedding": emb,
            "document": doc,
            "metadata": {"index": i}
        }
        for i, (emb, doc) in enumerate(zip(knowledge_embeddings, knowledge))
    ]

    await client.vectors.batch_upsert(
        project_id=project_id,
        vectors=vectors,
        namespace="knowledge"
    )

    # Query
    query = "What programming languages are good for AI?"
    query_emb = embeddings.embed_query(query)

    results = await client.vectors.search(
        project_id=project_id,
        query_vector=query_emb,
        namespace="knowledge",
        limit=3
    )

    print("\nSearch Results:")
    for r in results["results"]:
        print(f"- {r['document']}")
        print(f"  Score: {r['similarity_score']:.2f}\n")

asyncio.run(rag_example())

输出：

Search Results:
- Python is a versatile programming language used for web, AI, and data science
  Score: 0.87

- Machine learning is a subset of AI that learns from data
  Score: 0.72

- Neural networks are inspired by biological neurons
  Score: 0.65

📚 示例工作流程

常见用例的现成示例：

RAG工作流程 -使用免费嵌入构建生产RAG
代理内存 -持续对话历史
文件处理 -上传PDF、提取文本、搜索
NoSQL模式 -灵活的元数据存储
高级工作流 -多组件系统

🚀 快速入门指南

入门指南 -5分钟快速启动
故障排除 -常见问题和解决方案
迁移指南 -从松果体/Qdrant/Weaviate迁移
性能基准 -详细的性能指标

⚡ 演出

嵌入生成：约100条短信/秒（免费！）
向量搜索：对于100K矢量（使用HNSW），\<50ms
NoSQL查询：\<20ms（带索引）
文件上传：约500 KB/秒

看性能基准查看详细指标。

🔗 集成测试

完整的集成测试可在 tests/integration/test_zerodb_integration.py:

✅ RAG工作流程（单次+批量）
✅ 代理内存工作流
✅ 文件处理管道
✅ 多组件工作流
✅ 错误处理和边缘情况

运行测试：

cd tests/integration
ZERODB_API_KEY="your-key" pytest test_zerodb_integration.py -v

支持

文档：https://docs.ainative.studio/sdk/python
问题：https://github.com/ainative/zerodb-mcp-python/issues
电子邮件：support@ainative.studio
不一致：https://discord.gg/ainative

更新日志

1.0.0 (2025-01-14)

首次生产发布
60+MCP操作
完全异步支持
全面的错误处理
Pydantic的类型安全
90%+测试覆盖率

Zerodb Python SDK

快速接入

详细介绍