Python SDK

# Basic install
pip install crewlayer

# With integration extras
pip install "crewlayer[langchain]"
pip install "crewlayer[autogen]"
pip install "crewlayer[llamaindex]"

# Or all extras
pip install "crewlayer[all]"

# From source (for local dev)
git clone https://github.com/GerardSole/CrewLayer
cd CrewLayer && pip install -e ".[dev]"

from crewlayer import CrewLayerClient

# Explicit configuration
client = CrewLayerClient(
    base_url="http://localhost:8000",   # default: http://localhost:8000
    api_key="crwl_...",                 # or set CREWLAYER_API_KEY env var
    timeout=30.0,                       # default: 30s
)

# From environment variables (CREWLAYER_API_KEY, CREWLAYER_BASE_URL)
client = CrewLayerClient()

# Create
agent = client.agents.create(
    name="my-agent",
    description="Sales assistant",
    tags=["production", "sales"],
    config={"model": "claude-opus-4-8"},
)
agent_id = agent["id"]

# Get
agent = client.agents.get(agent_id)

# List (with optional filters)
agents = client.agents.list(status="working", tags="production")

# Update
agent = client.agents.update(
    agent_id,
    name="my-agent-v2",
    tags=["production", "sales", "v2"],
)

# Delete
client.agents.delete(agent_id)

# Status
status = client.agents.get_status(agent_id)
# → {"status": "idle", "current_session_id": None, "updated_at": "..."}

client.agents.set_status(agent_id, status="working", session_id="session-uuid")

# Status history
history = client.agents.status_history(agent_id, limit=20)

# Tags
client.agents.add_tags(agent_id, tags=["new-tag"])
client.agents.remove_tag(agent_id, tag="old-tag")

# Save a long-term memory
memory = client.memory.save(
    agent_id=agent_id,
    content="User is a Python developer who dislikes verbose code.",
    importance=0.9,        # 0.0–1.0
    source="manual",       # "manual" | "extracted" | "episode"
    session_id=session_id, # optional — link to session
)

# Semantic recall (cosine similarity search)
memories = client.memory.recall(
    agent_id=agent_id,
    query="what are the user's coding preferences?",
    top_k=5,               # default: 10
    min_similarity=0.7,    # optional threshold
)
for m in memories:
    print(f"[{m['similarity']:.2f}] {m['content']}")

# List all memories (paginated)
page = client.memory.list(agent_id=agent_id, limit=20)
memories = page["items"]
cursor   = page["next_cursor"]  # pass to next call for pagination

# Delete a memory
client.memory.delete(agent_id=agent_id, memory_id="mem-uuid")

# Stats
stats = client.memory.stats(agent_id=agent_id)
# → {"total": 42, "by_source": {...}, "avg_importance": 0.78}

# Short-term memory (Redis messages)
client.memory.append_message(
    agent_id=agent_id,
    session_id=session_id,
    role="user",            # "user" | "assistant" | "system"
    content="Hello, how are you?",
)
messages = client.memory.get_messages(agent_id=agent_id, session_id=session_id)
client.memory.clear_messages(agent_id=agent_id, session_id=session_id)

import time

# Log a tool call
start = time.monotonic()
result = my_tool(input_data)  # your tool call
duration_ms = int((time.monotonic() - start) * 1000)

action = client.actions.log(
    agent_id=agent_id,
    tool_name="web_search",
    input_params={"query": "latest AI news"},
    output_result={"results": result},
    status="success",       # "success" | "error" | "timeout"
    duration_ms=duration_ms,
    session_id=session_id,  # optional
)

# List actions (cursor-based pagination)
page = client.actions.list(
    agent_id=agent_id,
    tool="web_search",      # optional filter
    status="error",         # optional filter
    limit=50,
)
actions = page["items"]

# Get a single action
action = client.actions.get(agent_id=agent_id, action_id="action-uuid")

# Aggregated stats
stats = client.actions.stats(agent_id=agent_id)
# → {"total": 120, "error_rate": 0.05, "by_tool": {...}, "avg_duration_ms": 342}

# Create a new session at the start of a conversation
session = client.sessions.create(
    agent_id=agent_id,
    episode_id="episode-uuid",  # optional
    metadata={"user_id": "u_123"},
)
session_id = session["id"]

# Update agent status to working with this session
client.agents.set_status(agent_id, status="working", session_id=session_id)

# ... conversation happens ...

# Close the session — extracts memories from the conversation
result = client.sessions.close(session_id)
# Claude reads all messages and persists new long-term memories
print(f"Extracted {result['memories_extracted']} new memories")

# Reset agent status
client.agents.set_status(agent_id, status="idle")

# Write to the blackboard
entry = client.context.write(
    namespace="pipeline",
    key="step_1_output",
    value={"items": ["a", "b", "c"], "count": 3},
)
print(f"Version {entry['version']}")

# Read from the blackboard
entry = client.context.read(namespace="pipeline", key="step_1_output")
print(entry["value"])  # {"items": [...], "count": 3}

# List all keys in a namespace
keys = client.context.list_namespace(namespace="pipeline")

# Version history
history = client.context.history(namespace="pipeline", key="step_1_output")

# Rollback to a previous version
client.context.rollback(namespace="pipeline", key="step_1_output", version=2)

# Create an episode (a long-running task)
episode = client.episodes.create(
    agent_id=agent_id,
    title="Implement authentication feature",
    description="All sessions related to auth work",
)
episode_id = episode["id"]

# Recall memories scoped to this episode
memories = client.episodes.recall(
    agent_id=agent_id,
    episode_id=episode_id,
    query="what decisions were made about JWT expiry?",
    top_k=5,
)

# Complete when done
client.episodes.complete(agent_id=agent_id, episode_id=episode_id)

from crewlayer.integrations.langchain import CrewLayerMemory, CrewLayerCallbackHandler

# Drop-in memory for any LangChain chain
memory = CrewLayerMemory(client=client, agent_id=agent_id, session_id=session_id)

# Callback handler — auto-logs every chain step as a CrewLayer action
handler = CrewLayerCallbackHandler(client=client, agent_id=agent_id)

from langchain.chains import ConversationChain
chain = ConversationChain(memory=memory, callbacks=[handler])

from crewlayer.integrations.autogen import CrewLayerMemoryMixin

class MyAssistant(CrewLayerMemoryMixin, autogen.AssistantAgent):
    """Assistant with automatic memory recall and action logging."""
    pass

assistant = MyAssistant(
    name="assistant",
    crewlayer_client=client,
    crewlayer_agent_id=agent_id,
)