Agents

Agent integration: AI agents connect to Mixpeek via MCP, SDK, or REST to access ingest, search, classify, and monitor capabilities

Mixpeek exposes its entire platform as agent-callable tools. Connect via MCP for zero-code setup, use the built-in Agent Runtime for stateful conversations, or wire retrievers into LangChain, OpenAI, or any framework via REST.

Connect your agent to your own data

Create a namespace to get your API key and MCP endpoint, then give your agent search, ingest, and classify tools over your own content.

MCP (Model Context Protocol)

The fastest way to connect an AI agent to Mixpeek. Four hosted servers expose different tool scopes:

Scope	URL	Use it for
Full	`https://mcp.mixpeek.com/mcp`	Everything below in one server
Ingestion	`https://mcp.mixpeek.com/ingestion/mcp`	Buckets, uploads, collections, batches
Retrieval	`https://mcp.mixpeek.com/retrieval/mcp`	Retrievers, search, agent sessions
Admin	`https://mcp.mixpeek.com/admin/mcp`	Namespaces, keys, taxonomies, clusters

Claude Desktop
Claude Code
Cursor / Windsurf

{
  "mcpServers": {
    "mixpeek": {
      "url": "https://mcp.mixpeek.com/mcp",
      "headers": { "Authorization": "Bearer YOUR_API_KEY" }
    }
  }
}

claude mcp add mixpeek \
  --transport streamable-http \
  --url https://mcp.mixpeek.com/mcp \
  --header "Authorization: Bearer YOUR_API_KEY"

{
  "mcpServers": {
    "mixpeek": {
      "url": "https://mcp.mixpeek.com/mcp",
      "headers": { "Authorization": "Bearer YOUR_API_KEY" }
    }
  }
}

Per-Retriever Server

For a focused search agent, scope the MCP server to a single retriever. It reads your retriever’s input_schema and generates a typed search tool:

pip install mixpeek-mcp-retriever

mixpeek-mcp-retriever \
  --retriever-id ret_xxx \
  --namespace-id ns_xxx \
  --api-key YOUR_API_KEY

Exposes three tools: search (typed to your schema), describe (retriever metadata), and explain (pipeline walkthrough). Full MCP reference →

Agent Sessions

Mixpeek’s built-in agent runtime gives you stateful, multi-turn conversations backed by your data. Each session runs as a dedicated process with tool access, conversation memory, and SSE streaming.

# Create a session
curl -X POST "https://api.mixpeek.com/v1/agents/sessions" \
  -H "Authorization: Bearer $MIXPEEK_API_KEY" \
  -H "X-Namespace: $NAMESPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{
    "agent_config": {
      "system_prompt": "You help users search and analyze video content.",
      "available_tools": ["execute_retriever", "list_collections", "get_taxonomy"]
    }
  }'

# Send a message (SSE streaming)
curl -N -X POST "https://api.mixpeek.com/v1/agents/sessions/$SESSION_ID/messages" \
  -H "Authorization: Bearer $MIXPEEK_API_KEY" \
  -H "X-Namespace: $NAMESPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{ "content": "Find videos about machine learning", "stream": true }'

The agent reasons through a analyze → plan → execute → synthesize workflow, calling tools as needed and streaming events back:

Event	Description
`thinking`	Agent is analyzing or planning
`tool_call`	Agent is calling a tool
`tool_result`	Tool execution result
`message`	Response text chunk
`done`	Processing complete

Available Tools

Tool	Description
`execute_retriever`	Search documents via a retriever pipeline
`search_retrievers`	Find available retrievers
`get_retriever`	Get retriever configuration
`list_collections`	List collections in the namespace
`get_collection`	Get collection details
`list_taxonomies`	List taxonomies
`get_taxonomy`	Get taxonomy details
`list_clusters`	List cluster configurations
`get_object`	Get object metadata

Sessions persist for 7 days and automatically rehydrate after idle periods. Agent Sessions API →

LangChain

The langchain-mixpeek package provides a retriever, individual tools, and a full toolkit:

pip install langchain-mixpeek

from langchain_mixpeek import MixpeekToolkit
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic

toolkit = MixpeekToolkit(
    api_key="mxp_...",
    namespace="my-namespace",
    bucket_id="bkt_...",
    collection_id="col_...",
    retriever_id="ret_...",
)

agent = create_react_agent(
    ChatAnthropic(model="claude-sonnet-4-20250514"),
    toolkit.get_tools(),
)

result = agent.invoke({
    "messages": [("user", "Find product demos and summarize what's shown")]
})

Toolkit Tools

Tool	What it does
`mixpeek_search`	Search video, images, audio, documents
`mixpeek_ingest`	Upload content (text, images, video, audio, PDFs)
`mixpeek_process`	Trigger feature extraction
`mixpeek_classify`	Run taxonomy classification
`mixpeek_cluster`	Group similar documents
`mixpeek_alert`	Set up monitoring (webhook, Slack, email)

Scope tools to what your agent needs with toolkit.get_tools(actions=["search", "ingest"]). Full LangChain guide →

OpenAI Function Calling

Define a Mixpeek retriever as an OpenAI function schema:

from openai import OpenAI
from mixpeek import Mixpeek

openai_client = OpenAI()
mixpeek_client = Mixpeek(api_key="mxp_...")

tools = [{
    "type": "function",
    "function": {
        "name": "search_mixpeek",
        "description": "Search video, image, and audio content",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"},
                "limit": {"type": "integer", "description": "Max results"}
            },
            "required": ["query"]
        }
    }
}]

response = openai_client.chat.completions.create(
    model="gpt-4o", messages=messages, tools=tools
)

# Handle tool_calls by calling mixpeek_client.retrievers.execute()

Works with both the Chat Completions API and Assistants API. Full OpenAI guide →

Any Framework (REST)

The same pattern works with CrewAI, LlamaIndex, Haystack, Autogen, or plain HTTP — wrap the retriever execute endpoint as a tool:

import requests

def search(query: str, limit: int = 10) -> list:
    resp = requests.post(
        f"https://api.mixpeek.com/v1/retrievers/{RETRIEVER_ID}/execute",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "X-Namespace": NAMESPACE_ID,
            "Content-Type": "application/json",
        },
        json={"inputs": {"query_text": query}, "limit": limit},
    )
    return resp.json()["documents"]

Retriever Execute API →

Let an agent build the namespace

An agent can build a namespace end to end through the API: read the catalog, generate a manifest from a description, read its price, apply it, index files and grade the result. Give the agent an API key and this section. An agent given only this section and the billing page built a searchable namespace over 200 photos with the script below; its quote, its batch estimate and its charge were each $0.40.

Read the catalog

GET /v1/discovery returns the extractors, the models, the retriever stages and the manifest schema, with an example manifest and the authoring loop. GET /v1/public/templates/namespaces lists the templates.

Generate a manifest and its quote

POST /v1/manifest/generate takes description, manifest_name, and optionally inventory (files per bucket) and queries_per_month. The answer carries manifest (YAML), validation, suggested templates and models, and quote. When the manifest has one bucket and inventory has one entry, the quote prices that entry as the bucket and says so first in quote.assumptions. With several buckets, quote.error lists the bucket names. cost_estimate is a catalog figure at a stated volume (10,000 items and 100,000 queries a month); quote is the price for your files.

Edit the manifest

Read the YAML before applying it. A generated manifest can declare a storage connection and a bucket sync that need cloud credentials; remove them to upload files through the API. Add a field to the bucket schema and to the collection’s field_passthrough for anything you grade by, such as a file name. A retriever’s final_top_k caps every evaluation at that k, so raise it to at least twice your largest relevant set before grading. An edit can change the price, so quote the edited YAML again with POST /v1/organizations/billing/estimate; that quote is the one the batch is charged.

Validate, diff and dry-run

Send the YAML as the multipart file field manifest_file to POST /v1/manifest/validate, POST /v1/manifest/diff and POST /v1/manifest/apply?dry_run=true. The diff lists resources outside your manifest under in_system_only; apply leaves them alone, and the dry run’s would_create_count confirms it.

Apply, index and read the charge

POST /v1/manifest/apply returns each resource with its id. Each bucket also gets a system collection, mvs_bucket_<bucket_id>, which holds one document per object and costs nothing. Add objects, batch them, price the batch, submit it and poll it until its status is COMPLETED, COMPLETED_WITH_ERRORS, FAILED or CANCELED. The batch’s cost.cost_usd is what it was charged; GET /v1/organizations/billing/transactions lists the same charge.

Grade the retriever

Create a dataset, then start an evaluation with the metrics inside evaluation_config, and read it until its status is completed.

The whole loop in Python, with the shapes each call takes:

Python

import base64, os, pathlib, time, requests, yaml

API = "https://api.mixpeek.com"
H = {"Authorization": f"Bearer {os.environ['MIXPEEK_API_KEY']}"}
photos = sorted(pathlib.Path("photos").glob("*.jpg"))

# Generate, then quote the files you will index.
gen = requests.post(f"{API}/v1/manifest/generate", headers=H, json={
    "description": "search photos by the objects in them",
    "manifest_name": "photos",
    "inventory": {"photos": {"objects": len(photos), "images": len(photos)}},
    "queries_per_month": 10000,
}).json()
yaml_text = gen["manifest"]          # edit it here (step 3), then quote the edit:
bucket_name = yaml.safe_load(yaml_text)["buckets"][0]["name"]
quote = requests.post(f"{API}/v1/organizations/billing/estimate", headers=H, json={
    "manifest": yaml_text,
    "inventory": {bucket_name: {"objects": len(photos), "images": len(photos)}},
    "queries_per_month": 10000,
}).json()["manifest_quote"]
print("quote", quote["index_usd"], quote["monthly_usd"])
files = lambda: {"manifest_file": ("namespace.yaml", yaml_text, "application/x-yaml")}
assert requests.post(f"{API}/v1/manifest/validate", headers=H, files=files()).json()["valid"]
requests.post(f"{API}/v1/manifest/apply", headers=H, files=files(), params={"dry_run": "true"})
applied = requests.post(f"{API}/v1/manifest/apply", headers=H, files=files()).json()
ids = {r["resource_type"]: r["resource_id"] for r in applied["resources"]}
NS = {**H, "X-Namespace": ids["namespace"]}
bucket, retriever = ids["bucket"], ids["retriever"]

# Upload: one blob per object, plus the fields the bucket schema declares.
object_ids = []
for p in photos:
    data = "data:image/jpeg;base64," + base64.b64encode(p.read_bytes()).decode()
    obj = requests.post(f"{API}/v1/buckets/{bucket}/objects", headers=NS, json={
        "blobs": [{"property": "image", "type": "image", "data": data}],
        "file_name": p.name,
    }).json()
    object_ids.append(obj["object_id"])

batch = requests.post(f"{API}/v1/buckets/{bucket}/batches", headers=NS,
                      json={"object_ids": object_ids}).json()["batch_id"]
print(requests.post(f"{API}/v1/buckets/{bucket}/batches/{batch}/estimate-cost", headers=NS).json()["estimated_usd"])
requests.post(f"{API}/v1/buckets/{bucket}/batches/{batch}/submit", headers=NS, json={})
while (b := requests.get(f"{API}/v1/buckets/{bucket}/batches/{batch}", headers=NS).json())["status"] not in (
        "COMPLETED", "COMPLETED_WITH_ERRORS", "FAILED", "CANCELED"):
    time.sleep(5)
print(b["status"], b["cost"]["cost_usd"])

# Grade: relevant_documents are values of relevance_field, here file names.
requests.post(f"{API}/v1/retrievers/evaluations/datasets", headers=NS, json={
    "dataset_name": "photo-objects",
    "relevance_field": "file_name",
    "queries": [{"query_id": "handbag",
                 "query_input": {"query": "a photo with a handbag"},
                 "relevant_documents": ["000000001234.jpg", "..."]}],
})
ev = requests.post(f"{API}/v1/retrievers/{retriever}/evaluations", headers=NS, json={
    "dataset_name": "photo-objects",
    "evaluation_config": {"metrics": ["f1", "f2"], "k_multiples_of_relevant": [1, 2]},
}).json()["evaluation_id"]
while (e := requests.get(f"{API}/v1/retrievers/{retriever}/evaluations/{ev}", headers=NS).json())["status"] != "completed":
    time.sleep(5)
print(e["overall_metrics"]["f1_at_1r"], e["overall_metrics"]["f2_at_2r"])

query_input takes the retriever’s input fields (here its query input). f1_at_1r grades each query at k equal to its number of relevant documents, and f2_at_2r at twice that. On 200 COCO photos and five object queries, the run above scored F1 0.62 at k = R and F2 0.69 at k = 2R.

Choosing an Integration

I want to…	Use
Connect Claude or Cursor with no code	MCP
Build a stateful conversational agent	Agent Sessions
Build a LangChain/LangGraph agent	LangChain
Add tools to GPT models	OpenAI Function Calling
Use any other framework	REST
Have an agent build and price a namespace	Let an agent build the namespace

Get started

Connect your data

Extract features

Build retrievers

Enrich & organize

Integrate & operate

Resources

Connect your agent to your own data

MCP (Model Context Protocol)

Per-Retriever Server

Agent Sessions

Available Tools

LangChain

Toolkit Tools

OpenAI Function Calling

Any Framework (REST)

Let an agent build the namespace

Choosing an Integration

Connect your agent to your own data

​MCP (Model Context Protocol)

​Per-Retriever Server

​Agent Sessions

​Available Tools

​LangChain

​Toolkit Tools

​OpenAI Function Calling

​Any Framework (REST)

​Let an agent build the namespace

​Choosing an Integration

MCP (Model Context Protocol)

Per-Retriever Server

Agent Sessions

Available Tools

LangChain

Toolkit Tools

OpenAI Function Calling

Any Framework (REST)

Let an agent build the namespace

Choosing an Integration