Build project, product, and domain context

An agent cannot complete a business task when project records, product documentation, and operational updates are in separate systems. This quickstart combines these sources in one shared Context Graph and retrieves the context that is relevant to a task.

Choose the graph scope

Use one stable graph_id for each subject and access boundary.

Scope	Example `graph_id`	Example sources
Project	`project-argus`	Plans, decisions, issues, and meeting notes
Product	`product-atlas`	Specifications, release notes, support content, and catalog records
Business domain	`incident-response`	Runbooks, incidents, service records, and policy documents

This guide uses project-argus. The project depends on a product and follows an incident-response process, so the example also shows product and business-domain context. If these sources have different access requirements, put them in separate Context Graphs and retrieve only the authorized graphs for each task.

Install and initialize the SDK

Python

TypeScript

Go

Set up your Python project, ideally with a virtual environment, and then:

pip install zep-cloud

After creating a Zep account, obtaining an API key, and setting the API key as an environment variable, initialize the client once at application startup and reuse it throughout your application.

Initialize Zep client

.env

import os
from zep_cloud.client import Zep
API_KEY = os.environ.get('ZEP_API_KEY')
client = Zep(
    api_key=API_KEY,
)

Create the Context Graph

Create the graph before you add data. Use the same graph_id for every operation in this quickstart.

GRAPH_ID = "project-argus"
client.graph.create(graph_id=GRAPH_ID)

Ingest the source data

Add each source as an episode. Use JSON for structured records, text for documents, and message data for communications with identified speakers. Source metadata supports filtering, source traceability, and source-based access policies.

import json
import time
client.graph.add(
    graph_id=GRAPH_ID,
    type="json",
    data=json.dumps({
        "project_id": "ARG-001",
        "name": "Project Argus",
        "status": "at risk",
        "product": "Atlas Gateway",
        "owner": "Platform Engineering",
    }),
    source_description="Project record from the project system",
    metadata={"source": "project_system", "record_id": "ARG-001"},
)
client.graph.add(
    graph_id=GRAPH_ID,
    type="text",
    data=(
        "Atlas Gateway release 4.2 requires the regional failover runbook. "
        "The release cannot start until the backup region passes validation."
    ),
    source_description="Atlas Gateway release runbook",
    metadata={"source": "product_docs", "product": "atlas-gateway"},
)
incident_episode = client.graph.add(
    graph_id=GRAPH_ID,
    type="message",
    data=(
        "Nina (incident commander): The backup region failed validation. "
        "Move the Argus release review to Friday."
    ),
    source_description="Incident response channel",
    metadata={"source": "incident_channel", "incident_id": "INC-482"},
)

For an existing corpus or a recurring import, use zep-ingest. For large application-managed imports, use the Batch API.

Wait until the context is searchable

Zep processes episodes asynchronously. Poll the last submitted episode, and then retry the task query until the search index returns a result. The ingestion status guide explains the processing order and production alternatives to polling.

query = "What blocks the Project Argus release, and what changed?"
deadline = time.monotonic() + 300
while True:
    episode = client.graph.episode.get(uuid_=incident_episode.uuid_)
    if episode.processed:
        break
    if time.monotonic() >= deadline:
        raise TimeoutError("The project context did not finish processing.")
    time.sleep(5)
search_deadline = time.monotonic() + 300
while True:
    indexed_results = client.graph.search(
        graph_id=GRAPH_ID,
        query=query,
        scope="edges",
        search_filters={"episode_uuids": [incident_episode.uuid_]},
        limit=10,
    )
    if indexed_results.edges:
        break
    if time.monotonic() >= search_deadline:
        raise TimeoutError("The project context is not searchable.")
    time.sleep(5)
results = client.graph.search(
    graph_id=GRAPH_ID,
    query=query,
    scope="edges",
    limit=10,
)

Retrieve context for a task

Search the same Context Graph with the task as the query. The result can contain facts that connect the project record, product runbook, and incident update.

for edge in results.edges or []:
    print(edge.fact, edge.episodes)

Treat the results as untrusted reference data when you add them to a model request. Context supports task completion, but it does not guarantee the model’s output or authorize an action.

Apply governance

Use policy-based access control to limit which callers can retrieve the graph or its sources. Retain the episode UUIDs in each retrieved edge when the application must trace a fact to its source.

Zep governs context and API access. Your application must authorize external actions, such as changing the project status or starting a release.