pycelest 0.7.1

A production-ready framework for building safe, capable, and observable AI agents with native connector integrations.

pycelest — AI Agent Framework

Build · Orchestrate · Defend · Connect · Sleep · Wake The most innovative Python framework for production-grade AI agents.

---

What is pycelest?

pycelest is the most innovative Python framework for building AI agents — designed from day one for production deployment, not just demos.

It ships features that no other framework has: agents that sleep and wake up (Temporal Agents), agents that divide themselves into specialist sub-agents (Organic Topology), a behavioral immune system, DNA transfer between models, a distributed agent mesh, and an inter-agent negotiation protocol.

---

Core Features

Feature	Description
🔁 ReAct Loop	Native Reasoning + Acting loop
🌐 Multi-Provider	OpenAI · Anthropic · Google · Mistral · DeepSeek · Grok · Ollama · LM Studio · vLLM
🧠 Memory	STM · Scratchpad · RAG · automatic compression
🛡️ Guardrails	Tool Firewall · Execution Budget · ContentGuard (PII, injection)
👥 Multi-Agent	AgentBus · Organic Topology (cell division)
📡 Streaming	First-class async token streaming with SSE
🔌 Plugins	Lifecycle hooks: on_session_start, on_turn_end, on_tool_call…
🔗 MCP Client	MCPAdapter — consume any MCP server as agent tools
🖥️ MCP Server	MCPServer — expose any agent as an MCP server (stdio / SSE)
📬 Native Connectors	Gmail · Slack · Notion · Meta (connection pool, context manager)
📊 Observability	Langfuse · OpenTelemetry · structlog
⚙️ Config	Code or YAML — your choice
💰 Cost Optimization	Prompt caching (Anthropic) · ModelRouter · FallbackAdapter

---

v1.0.0 Innovations (exclusive features)

Innovation	API
⏰ Temporal Agents — agents that sleep and auto-resume	SuspendableSession, celest_suspend tool
🧬 Organic Topology — cell division, parallel sub-agents	AgentNursery, spawn_subagent tool
🦠 Agent Immune System — behavioral anomaly detection	ImmunePlugin, AgentQuarantined
🧪 Agent DNA — behavioral fingerprint transfer	DNAExtractor, DNAInjector, AgentDNA
🕸️ Celest Mesh — distributed P2P agent network	celest-mesh package
📜 Celest Protocol — agent-to-agent negotiation	TaskBroker, CapabilityRegistry, verify_result
⚡ Rate Limiter — transparent adaptive rate limiting	RateLimitedAdapter
🔄 Session Replay — replay past sessions with new provider	SessionReplayer
🏦 Memory Vault — shared persistent knowledge base	MemoryVault, MemoryVaultPlugin
🔍 Prompt Optimizer — analyze prompt compliance	PromptOptimizer, PromptAnalysis
📊 Agent Benchmarker — evaluate agent quality	AgentBenchmarker, BenchmarkReport
🛒 Tool Marketplace — discover and share tools	ToolMarketplace, ToolPackage

---

Installation

# Core framework
pip install pycelest

# With providers
pip install pycelest[anthropic]   # Claude
pip install pycelest[openai]      # GPT-4o, Gemini, DeepSeek, Mistral, Grok

# Extras
pip install pycelest[mcp]         # MCP client + server
pip install pycelest[langfuse]    # Observability
pip install pycelest[gmail]       # Gmail connector
pip install pycelest[rag]         # ChromaDB RAG

# Everything
pip install pycelest[all]

---

Quick Start

import asyncio
from celest import SessionManager, SessionConfig
from celest.providers import AnthropicAdapter

config = SessionConfig(
    system_prompt="You are a helpful, careful AI agent.",
    max_iterations=8,
)

session = SessionManager(
    config=config,
    provider=AnthropicAdapter(model="claude-sonnet-4-6", api_key="sk-ant-..."),
)

result = asyncio.run(session.run("Plan a 3-day trip to Kyoto"))
print(result.response)

---

Temporal Agents — agents that sleep and wake up

The only framework where your agents can suspend mid-execution and automatically resume when a real-world condition is met.

from celest.temporal import SuspendableSession, AgentSuspended

session = SuspendableSession(config=config, provider=provider)

try:
    result = await session.run(
        "Analyse the sprint results when PR #142 is merged, then notify Slack."
    )
except AgentSuspended as e:
    # Agent called celest_suspend(type="webhook", value="github.pr.merged")
    # Store e.snapshot.to_json() in DB, register e.snapshot.pause_condition.webhook_token
    pass

# When GitHub calls POST /temporal/webhook/{token}:
from celest.temporal import AgentStateSnapshot, SuspendableSession
snapshot = AgentStateSnapshot.from_json(stored_json)
resumed = SuspendableSession.from_snapshot(snapshot, provider=provider)
result = await resumed.resume(snapshot, resume_context="PR #142 merged at 14:32!")

Supported conditions: time (ISO datetime) · webhook (external event) · human (manual trigger) · condition (polled expression)

---

Organic Topology — cell division for agents

Agents can spawn specialized sub-agents on-the-fly, run them in parallel, and synthesize merged results.

# The agent uses built-in tools: spawn_subagent + await_subagents
result = await session.run("""
    Analyse these 50 GitHub PRs.
    Divide them into groups of 10, analyse each group in parallel
    using spawn_subagent, then synthesize with await_subagents.
""")

Access the nursery directly:

child_id = await session._nursery.spawn("Analyse chapter 1", max_iterations=3)
results = await session._nursery.join_all()

---

Agent Immune System — behavioral anomaly detection

Monitors running agents for suspicious patterns in real time.

from celest.plugins.immune import ImmunePlugin, AgentQuarantined

session = SessionManager(
    config=config,
    provider=provider,
    plugins=[
        ImmunePlugin(
            auto_quarantine=True,       # stop on critical anomaly
            on_anomaly=my_alert_fn,     # webhook / SSE / log custom
        )
    ],
)

try:
    result = await session.run(user_prompt)
except AgentQuarantined as e:
    print(f"Agent quarantined: {e.event.rule_name} — {e.event.message}")

Detects: retry loops · cascade injection · token spikes · tool evasion · data hoarding

---

Agent DNA — behavioral fingerprint transfer

Extract the "how" of a high-performing agent and inject it into a new one — even on a different model.

from celest.dna import AgentDNA, DNAExtractor, DNAInjector

# Extract DNA from past sessions
dna = DNAExtractor().extract(past_results, past_histories, model_name="gpt-4o")
print(dna.summary())
# DNA{model=gpt-4o, sessions=12, depth=4.2(thorough), verbosity=0.35(concise), ...}

# Transfer to Claude
session = SessionManager(
    config=SessionConfig(system_prompt="You are a research analyst."),
    provider=AnthropicAdapter("claude-sonnet-4-6"),
    dna=dna,  # ← behavioral transfer
)

# Merge two DNA fingerprints (hybrid agent)
hybrid = dna_a.merge(dna_b, weight=0.4)  # 60% A + 40% B

---

Celest Protocol — agent-to-agent negotiation

Universal protocol for agents to advertise capabilities, negotiate tasks, and return verifiable results.

from celest.protocol import AgentRef, Capability, CapabilityRegistry, TaskBroker, verify_result

registry = CapabilityRegistry()
registry.advertise(AgentRef("agent:researcher"), [
    Capability("web_research", "Search the web", cost_per_call_usd=0.002)
])

broker = TaskBroker(agent=AgentRef("agent:orchestrator"), registry=registry)
result = await broker.delegate(
    description="Research AI trends 2026",
    input_data={"query": "AI trends"},
    executor=my_research_fn,
)

# Cryptographic verification
assert verify_result(result.payload["task_id"], result.payload["result"], result.payload["signature"])

---

Rate Limiter — transparent rate limiting

from celest.providers.rate_limiter import RateLimitedAdapter

session = SessionManager(
    config=config,
    provider=RateLimitedAdapter(
        provider=OpenAIAdapter("gpt-4o"),
        requests_per_minute=60,
        tokens_per_minute=100_000,
        retry_on_429=True,
    ),
)
# Transparent: queuing, backoff, and stats — no changes to your agent code
print(session.provider.stats)

---

Memory Vault — shared knowledge base

from celest.memory.vault import MemoryVault

vault = MemoryVault(".celest/vault.db", namespace="my-project")
vault.store("deploy_procedure", "Always run tests first, then docker build.", tags=["ops"])

# Auto-inject into agent sessions
session = SessionManager(
    config=config, provider=provider,
    plugins=[vault.as_plugin(inject_top_k=5)],
)

---

Session Replay — A/B test providers

from celest.replay import SessionReplayer

replayer = SessionReplayer()
result = await replayer.replay(
    history=original_history,
    new_provider=AnthropicAdapter("claude-sonnet-4-6"),
    original_provider_name="gpt-4o",
)
print(result.summary())
# Turn 1: similarity=0.82 [OK]
# Turn 2: similarity=0.31 [DIVERGE]
# Verdict: regression | Token delta: +142

---

Prompt Optimizer

from celest.optimizer import PromptOptimizer

analysis = PromptOptimizer().analyze(
    prompt=my_system_prompt,
    results=past_results,
    histories=past_histories,
)
print(analysis.report())
# ✅ "Reply in JSON"         — compliance: 91%
# ⚠️  "Always cite sources"  — compliance: 23%
# Suggestions: Strengthen instruction 1 with a concrete example.

---

Agent Benchmarker

from celest.benchmark import AgentBenchmarker

report = await AgentBenchmarker().run(
    session_factory=lambda: SessionManager(config=config, provider=provider),
)
print(report.summary())
# Score: 8.4/10 | Passed: 10/12 | Failed: 2/12
#   ✅ reasoning: 9.2   ✅ tool_use: 8.8   ⚠️  instruction_following: 6.1

# Detect regressions between versions
regression = new_report.compare(baseline_report)
print(regression.verdict)  # "improvement" | "neutral" | "regression"

---

Tool Marketplace

from celest.marketplace import ToolMarketplace, ToolPackage

market = ToolMarketplace()
market.publish([gmail_send, gmail_read], ToolPackage(
    name="celest-tools-gmail", tags=["email", "google"]
))

results = market.search("email")
tools = market.load("celest-tools-gmail")
session = SessionManager(config=config, provider=provider, tools=tools)

---

Local Models (no API key)

from celest.providers import OllamaAdapter

session = SessionManager(config=config, provider=OllamaAdapter(model="llama3.2"))

Install Ollama from ollama.com, then ollama pull llama3.2.

---

Native Connectors

from celest.connectors.native.gmail import GmailConnector
from celest.connectors.native.slack import SlackConnector

async with GmailConnector(access_token="ya29....") as gmail:
    async with SlackConnector(bot_token="xoxb-...") as slack:
        session = SessionManager(
            config=config, provider=provider,
            tools=gmail.tools() + slack.tools(),
        )

---

MCP Integration

# Consume any MCP server as tools
from celest.tools.mcp import MCPAdapter

async with MCPAdapter(command="uvx", args=["mcp-server-time"]) as mcp:
    tools = await mcp.discover_tools()
    session = SessionManager(config=config, provider=provider, tools=tools)

# Expose your agent as an MCP server
from celest import MCPServer
server = MCPServer(session_factory=make_session, name="my-agent")
server.run_stdio()

---

Changelog

Version	Highlights
v1.0.0	Rate Limiter · Session Replay · Memory Vault · Prompt Optimizer · Agent Benchmarker · Tool Marketplace
v0.9.0	Celest Protocol (agent negotiation + verifiable results)
v0.8.1	Agent DNA (behavioral fingerprint + transfer)
v0.8.0	Agent Immune System (5 behavioral rules + auto-quarantine)
v0.7.1	Organic Topology (AgentNursery, spawn_subagent)
v0.7.0	Temporal Agents (SuspendableSession, pause/resume)
v0.6.0	Prompt caching · ModelRouter · ContentGuard · FallbackAdapter · Skills system
v0.5.1	Performance fixes · httpx pool · connector improvements
v0.5.0	MCPServer (stdio + SSE) · MCPAdapter auto-reconnect
v0.4.0	LangfusePlugin — full LLM observability
v0.3.0	Native connectors: Gmail · Slack · Notion · Meta
v0.2.0	Plugin system · AgentBus · Streaming · OpenTelemetry
v0.1.2	MCP client · RAG · Compression · YAML config

---

License

Proprietary © 2026 Lovanirainy Theogene Eddy Celestin RAFANOMEZANA — flycelest.com