refactor: update consumer, admin, exceptions, and __init__ for Moonshot API

Author: josedab

examples/agent_memory/memory_demo.py

@@ -0,0 +1,125 @@
+"""Streamline Agent Memory Example.
+
+Demonstrates the memory MCP tools (remember, recall) for building
+agents with persistent, semantically searchable memory. Shows both
+single-agent memory and multi-agent shared memory.
+
+Prerequisites:
+    - Streamline server running with memory features enabled
+    - pip install streamline-sdk
+
+Run with:
+    python examples/agent_memory/memory_demo.py
+"""
+
+import asyncio
+import os
+
+from streamline_sdk import StreamlineClient
+
+
+async def single_agent_memory(client: StreamlineClient) -> None:
+    """Demonstrate remember/recall for a single agent."""
+    print("=== Single Agent Memory ===")
+
+    # Store architectural decisions
+    await client.memory_remember(
+        agent_id="demo-agent",
+        content="We chose PostgreSQL for its JSONB support and mature ecosystem",
+        kind="fact",
+        importance=0.8,
+        tags=["architecture", "database"],
+    )
+
+    await client.memory_remember(
+        agent_id="demo-agent",
+        content="Redis is used as a caching layer with a 15-minute TTL",
+        kind="fact",
+        importance=0.7,
+        tags=["architecture", "caching"],
+    )
+
+    await client.memory_remember(
+        agent_id="demo-agent",
+        content="User requested dark mode support in the dashboard",
+        kind="preference",
+        importance=0.6,
+        tags=["ui", "user-request"],
+    )
+
+    print("Stored 3 memories\n")
+
+    # Recall by semantic similarity
+    print("--- Recall: 'why did we pick our database?' ---")
+    results = await client.memory_recall(
+        agent_id="demo-agent",
+        query="why did we pick our database?",
+        k=5,
+    )
+    for hit in results:
+        print(f"  [{hit.tier}] score={hit.score:.2f}: {hit.content}")
+
+    print("\n--- Recall: 'caching strategy' ---")
+    results = await client.memory_recall(
+        agent_id="demo-agent",
+        query="caching strategy",
+        k=5,
+    )
+    for hit in results:
+        print(f"  [{hit.tier}] score={hit.score:.2f}: {hit.content}")
+
+
+async def multi_agent_shared_memory(client: StreamlineClient) -> None:
+    """Demonstrate shared memory between multiple agents."""
+    print("\n=== Multi-Agent Shared Memory ===")
+
+    # Agent A stores a decision
+    await client.memory_remember(
+        agent_id="agent-a",
+        namespace="team-shared",
+        content="Deploy target is Kubernetes on AWS EKS",
+        kind="fact",
+        importance=0.9,
+        tags=["infra", "deployment"],
+    )
+    print("Agent A stored deployment decision")
+
+    # Agent B stores related context
+    await client.memory_remember(
+        agent_id="agent-b",
+        namespace="team-shared",
+        content="CI/CD pipeline uses GitHub Actions with OIDC auth to AWS",
+        kind="fact",
+        importance=0.8,
+        tags=["infra", "ci-cd"],
+    )
+    print("Agent B stored CI/CD context")
+
+    # Agent C recalls shared memories from the team namespace
+    print("\n--- Agent C recalls 'deployment infrastructure' from shared namespace ---")
+    results = await client.memory_recall(
+        agent_id="agent-c",
+        namespace="team-shared",
+        query="deployment infrastructure",
+        k=5,
+    )
+    for hit in results:
+        print(f"  [{hit.tier}] score={hit.score:.2f}: {hit.content}")
+
+
+async def main() -> None:
+    """Run agent memory demos."""
+    bootstrap = os.environ.get("STREAMLINE_BOOTSTRAP_SERVERS", "localhost:9092")
+    http_url = os.environ.get("STREAMLINE_HTTP", "http://localhost:9094")
+
+    async with StreamlineClient(
+        bootstrap_servers=bootstrap, http_endpoint=http_url
+    ) as client:
+        await single_agent_memory(client)
+        await multi_agent_shared_memory(client)
+
+    print("\nDone!")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

streamline_sdk/__init__.py

@@ -19,9 +19,9 @@ async def main():
 
 from .client import StreamlineClient
 from .producer import Producer, ProducerRecord, RecordMetadata
-from .consumer import Consumer, ConsumerRecord
+from .consumer import Consumer, ConsumerRecord, SearchHit
 from .admin import Admin, TopicConfig, TopicInfo, PartitionInfo
-from .admin import ClusterInfo, BrokerInfo, ConsumerLag, ConsumerGroupLag, InspectedMessage, MetricPoint
+from .admin import ClusterInfo, BrokerInfo, ConsumerLag, ConsumerGroupLag, InspectedMessage, MetricPoint, BranchInfo
 from .exceptions import (
     StreamlineError,
     ConnectionError,
@@ -43,6 +43,40 @@ async def main():
 )
 from .schema_producer import SchemaProducer, SchemaConsumer, DeserializedRecord
 from .traced import TracedProducer, TracedConsumer
+from .branches_admin import (
+    BranchAdminClient,
+    BranchAdminError,
+    BranchMessage,
+    BranchView,
+)
+from .contracts import (
+    ContractsClient,
+    ContractsError,
+    ValidationError,
+    ValidationResult,
+)
+from .attestation import (
+    ATTEST_HEADER,
+    Attestor,
+    AttestationError,
+    SignedAttestation,
+)
+from .verifier import (
+    StreamlineVerifier,
+    VerificationResult as AttestationVerificationResult,
+)
+from .search import (
+    SearchClient,
+    SearchError,
+    SearchHit,
+    SearchResult,
+)
+from .memory import (
+    MemoryClient,
+    MemoryError,
+    RecalledMemory,
+    WrittenEntry,
+)
 
 __version__ = "0.2.0"
 
@@ -61,6 +95,7 @@ async def main():
     "TopicConfig",
     "TopicInfo",
     "PartitionInfo",
+    "BranchInfo",
     # Exceptions
     "StreamlineError",
     "ConnectionError",
@@ -98,4 +133,32 @@ async def main():
     # Traced wrappers
     "TracedProducer",
     "TracedConsumer",
+    # Branches admin (M5 P1)
+    "BranchAdminClient",
+    "BranchAdminError",
+    "BranchMessage",
+    "BranchView",
+    # Contracts validate (M2)
+    "ContractsClient",
+    "ContractsError",
+    "ValidationError",
+    "ValidationResult",
+    # Attestation (M4)
+    "ATTEST_HEADER",
+    "Attestor",
+    "AttestationError",
+    "SignedAttestation",
+    # Local attestation verifier
+    "StreamlineVerifier",
+    "AttestationVerificationResult",
+    # Semantic search (M2)
+    "SearchClient",
+    "SearchError",
+    "SearchHit",
+    "SearchResult",
+    # Agent Memory (M1)
+    "MemoryClient",
+    "MemoryError",
+    "RecalledMemory",
+    "WrittenEntry",
 ]

streamline_sdk/admin.py

@@ -208,6 +208,23 @@ class MetricPoint:
     timestamp: int = 0
 
 
+@dataclass
+class BranchInfo:
+    """Information about a copy-on-write topic branch (M5).
+
+    Attributes:
+        name: Branch name.
+        base_topic: The base topic this branch forks from.
+        state: Branch state (active, discarded, merged).
+        created_at: Creation timestamp in milliseconds since epoch.
+    """
+
+    name: str
+    base_topic: str
+    state: str = "active"
+    created_at: int = 0
+
+
 class Admin:
     """Administrative operations for Streamline.
 
@@ -399,6 +416,51 @@ def _sync_get():
                     return json.loads(resp.read())
             return await asyncio.to_thread(_sync_get)
 
+    async def _http_post(self, path: str, body: Any) -> Any:
+        """Make an HTTP POST request to the Streamline REST API."""
+        http_url = getattr(self._client_config, "http_url", "http://localhost:9094")
+        url = f"{http_url}{path}"
+
+        if HAS_AIOHTTP:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(url, json=body, timeout=aiohttp.ClientTimeout(total=10)) as resp:
+                    if resp.status == 404:
+                        raise TopicError(f"Not found: {path}")
+                    if resp.status not in (200, 201):
+                        text = await resp.text()
+                        raise TopicError(f"HTTP {resp.status}: {text}")
+                    return await resp.json()
+        else:
+            import urllib.request
+            payload = json.dumps(body).encode("utf-8")
+            req = urllib.request.Request(url, data=payload, method="POST")
+            req.add_header("Content-Type", "application/json")
+            def _sync_post():
+                with urllib.request.urlopen(req, timeout=10) as resp:
+                    return json.loads(resp.read())
+            return await asyncio.to_thread(_sync_post)
+
+    async def _http_delete(self, path: str) -> None:
+        """Make an HTTP DELETE request to the Streamline REST API."""
+        http_url = getattr(self._client_config, "http_url", "http://localhost:9094")
+        url = f"{http_url}{path}"
+
+        if HAS_AIOHTTP:
+            async with aiohttp.ClientSession() as session:
+                async with session.delete(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
+                    if resp.status == 404:
+                        raise TopicError(f"Not found: {path}")
+                    if resp.status >= 300:
+                        text = await resp.text()
+                        raise TopicError(f"HTTP {resp.status}: {text}")
+        else:
+            import urllib.request
+            req = urllib.request.Request(url, method="DELETE")
+            def _sync_delete():
+                with urllib.request.urlopen(req, timeout=10) as resp:
+                    pass
+            await asyncio.to_thread(_sync_delete)
+
     async def list_consumer_groups(self) -> List[str]:
         """List all consumer groups.
 
@@ -609,6 +671,62 @@ async def metrics_history(self) -> List[MetricPoint]:
             for m in data
         ]
 
+    async def create_branch(
+        self, name: str, base_topic: str, base_offsets: Optional[Dict[int, int]] = None
+    ) -> "BranchInfo":
+        """Create a copy-on-write branch of a topic (M5).
+
+        Args:
+            name: Branch name.
+            base_topic: Topic to branch from.
+            base_offsets: Per-partition base offsets (partition -> offset).
+
+        Returns:
+            BranchInfo for the newly created branch.
+        """
+        body: Dict[str, Any] = {"name": name, "base_topic": base_topic}
+        if base_offsets:
+            body["base_offsets"] = base_offsets
+        data = await self._http_post("/v1/branches", body)
+        return BranchInfo(
+            name=data.get("name", name),
+            base_topic=data.get("base_topic", base_topic),
+            state=data.get("state", "active"),
+            created_at=int(data.get("created_at", 0)),
+        )
+
+    async def list_branches(self, topic: Optional[str] = None) -> List["BranchInfo"]:
+        """List copy-on-write topic branches (M5).
+
+        Args:
+            topic: Filter by base topic (optional).
+
+        Returns:
+            List of BranchInfo objects.
+        """
+        path = "/v1/branches"
+        if topic:
+            path += f"?topic={topic}"
+        data = await self._http_get(path)
+        items = data if isinstance(data, list) else data.get("items", [])
+        return [
+            BranchInfo(
+                name=b.get("name", ""),
+                base_topic=b.get("base_topic", ""),
+                state=b.get("state", "active"),
+                created_at=int(b.get("created_at", 0)),
+            )
+            for b in items
+        ]
+
+    async def discard_branch(self, branch_id: str) -> None:
+        """Discard (delete) a copy-on-write topic branch (M5).
+
+        Args:
+            branch_id: Branch identifier.
+        """
+        await self._http_delete(f"/v1/branches/{branch_id}")
+
     async def __aenter__(self) -> "Admin":
         """Enter async context manager."""
         await self.start()