Publishers

from tenxgraph.core.graph import StateGraph
from tenxgraph.runtime.publisher import BasePublisher, ConsolePublisher
from tenxgraph.runtime.publisher.events import Event, EventType, ContentType, EventModel
from tenxgraph.runtime.publisher import publish_event

# Optional backends
from tenxgraph.runtime.publisher import RedisPublisher     # pip install "10xgraph[redis]"
from tenxgraph.runtime.publisher import KafkaPublisher     # pip install "10xgraph[kafka]"
from tenxgraph.runtime.publisher import RabbitMQPublisher  # pip install "10xgraph[rabbitmq]"

# Fan-out
from tenxgraph.runtime.publisher import CompositePublisher

# Tracing backends
from tenxgraph.runtime.publisher import (
    ObservabilityLevel,
    OtelPublisher,
    LogfirePublisher,
    LangsmithPublisher,
    setup_tracing,
    setup_logfire,
    setup_langsmith,
    setup_observability,
)

class EventModel(BaseModel):
    event: Event
    event_type: EventType
    content: str = ""
    content_blocks: list[ContentBlock] | None = None
    data: dict[str, Any] = {}
    content_type: list[ContentType] | None = None
    node_name: str = ""
    run_id: str  # default: new UUID4 string
    thread_id: str | int = ""
    user_id: str | int | None = None
    timestamp: float  # default: current time
    is_error: bool = False
    metadata: dict[str, Any] = {}

from tenxgraph.runtime.publisher.events import EventModel, Event, EventType, ContentType

# The runtime normally creates these; build one by hand for custom events
event = EventModel(
    event=Event.NODE_EXECUTION,
    event_type=EventType.START,
    node_name="agent",
    run_id="run-123",
    thread_id="thread-456",
    content_type=[ContentType.MESSAGE],
    data={"model": "gpt-4", "temperature": 0.7},
)

class BasePublisher(ABC):
    def __init__(self, config: dict[str, Any])

    @abstractmethod
    async def publish(self, event: EventModel) -> Any
    
    @abstractmethod
    async def close(self) -> None

    @abstractmethod
    def sync_close(self) -> None

import asyncio

from tenxgraph.core.graph import StateGraph
from tenxgraph.core.state import AgentState, Message
from tenxgraph.runtime.publisher import ConsolePublisher
from tenxgraph.utils.constants import END, START


def echo(state: AgentState) -> list[Message]:
    return [Message.text_message("hello", role="assistant")]


async def main() -> None:
    # The publisher is closed automatically when the block exits
    async with ConsolePublisher() as publisher:
        graph = StateGraph(publisher=publisher)
        graph.add_node("echo", echo)
        graph.add_edge(START, "echo")
        graph.add_edge("echo", END)
        app = graph.compile()
        await app.ainvoke({"messages": [Message.text_message("hi")]})


asyncio.run(main())

class ConsolePublisher(BasePublisher):
    def __init__(self, config: dict[str, Any] | None = None)

from tenxgraph.core.graph import StateGraph
from tenxgraph.runtime.publisher import ConsolePublisher

# Print events to stdout
publisher = ConsolePublisher()

# Or route them through the logger, so they follow your logging config
publisher = ConsolePublisher(config={"use_logger": True})

graph = StateGraph(publisher=publisher)

class RedisPublisher(BasePublisher):
    def __init__(self, config: dict[str, Any] | None = None)

from tenxgraph.core.graph import StateGraph
from tenxgraph.runtime.publisher import RedisPublisher

# Pub/Sub on a named channel
pubsub = RedisPublisher(config={"url": "redis://localhost:6379/0", "channel": "agent-events"})

# Stream mode with a bounded length
stream = RedisPublisher(config={"mode": "stream", "stream": "agent-events", "maxlen": 10000})

graph = StateGraph(publisher=stream)

class KafkaPublisher(BasePublisher):
    def __init__(self, config: dict[str, Any] | None = None)

from tenxgraph.core.graph import StateGraph
from tenxgraph.runtime.publisher import KafkaPublisher

publisher = KafkaPublisher(config={
    "bootstrap_servers": "localhost:9092",
    "topic": "agent-events",
    "compression_type": "gzip",
})

graph = StateGraph(publisher=publisher)

class RabbitMQPublisher(BasePublisher):
    def __init__(self, config: dict[str, Any] | None = None)

from tenxgraph.core.graph import StateGraph
from tenxgraph.runtime.publisher import RabbitMQPublisher

publisher = RabbitMQPublisher(config={
    "url": "amqp://guest:guest@localhost:5672/",
    "exchange": "agent-events",
    "routing_key": "graph.events",
})

graph = StateGraph(publisher=publisher)

class CompositePublisher(BasePublisher):
    def __init__(self, publishers: list[BasePublisher])
    def add_publisher(self, publisher: BasePublisher) -> None
    def remove_publisher(self, publisher: BasePublisher) -> None

from tenxgraph.core.graph import StateGraph
from tenxgraph.runtime.publisher import CompositePublisher, ConsolePublisher, RedisPublisher

publisher = CompositePublisher([
    ConsolePublisher(),
    RedisPublisher({"url": "redis://localhost:6379/0"}),
])

graph = StateGraph(publisher=publisher)
# Equivalent: StateGraph(publisher=[ConsolePublisher(), RedisPublisher(...)])

class OtelPublisher(BasePublisher):
    def __init__(
        self,
        tracer: Tracer | None = None,
        level: ObservabilityLevel = ObservabilityLevel.STANDARD,
    )

def setup_tracing(
    graph: StateGraph,
    tracer: Tracer | None = None,
    level: ObservabilityLevel = ObservabilityLevel.STANDARD,
) -> OtelPublisher

from tenxgraph.core.graph import StateGraph
from tenxgraph.runtime.publisher import ObservabilityLevel, setup_tracing

graph = StateGraph()
# ... add nodes and edges

# Attach tracing before compile
setup_tracing(graph, level=ObservabilityLevel.STANDARD)
app = graph.compile()

class LogfirePublisher(OtelPublisher):
    def __init__(
        self,
        *,
        token: str | None = None,
        service_name: str | None = None,
        send_to_logfire: bool = True,
        console: Any = None,
        level: ObservabilityLevel = ObservabilityLevel.STANDARD,
        additional_span_processors: list | None = None,
        **configure_kwargs: Any,
    )

def setup_logfire(
    graph: StateGraph | None = None,
    *,
    token: str | None = None,
    service_name: str | None = None,
    send_to_logfire: bool = True,
    console: Any = None,
    level: ObservabilityLevel = ObservabilityLevel.STANDARD,
    additional_span_processors: list | None = None,
    **configure_kwargs: Any,
) -> None

from tenxgraph.core.graph import StateGraph
from tenxgraph.runtime.publisher import setup_logfire

graph = StateGraph()
# ... add nodes and edges

# Reads LOGFIRE_TOKEN from the environment
setup_logfire(graph, service_name="my-agent")
app = graph.compile()

class LangsmithPublisher(OtelPublisher):
    def __init__(
        self,
        *,
        api_key: str | None = None,
        project: str | None = None,
        endpoint: str = "https://api.smith.langchain.com/otel",
        level: ObservabilityLevel = ObservabilityLevel.STANDARD,
        tracer_provider: Any = None,
    )

def setup_langsmith(
    graph: StateGraph | None = None,
    *,
    api_key: str | None = None,
    project: str | None = None,
    endpoint: str = "https://api.smith.langchain.com/otel",
    level: ObservabilityLevel = ObservabilityLevel.STANDARD,
    tracer_provider: Any = None,
) -> None

from tenxgraph.core.graph import StateGraph
from tenxgraph.runtime.publisher import setup_langsmith

graph = StateGraph()
# ... add nodes and edges

# Reads LANGSMITH_API_KEY from the environment
setup_langsmith(graph, project="my-project")
app = graph.compile()

def setup_observability(graph: StateGraph | None, config: dict[str, Any]) -> None

from tenxgraph.core.graph import StateGraph
from tenxgraph.runtime.publisher import setup_observability

graph = StateGraph()
# ... add nodes and edges

setup_observability(graph, {
    "level": "standard",
    "logfire": {"enabled": True, "service_name": "my-agent"},
    "langsmith": {"enabled": True, "project": "eval"},
})
app = graph.compile()

def publish_event(
    event: EventModel,
    publisher: BasePublisher | None = Inject[BasePublisher],
    task_manager: BackgroundTaskManager = Inject[BackgroundTaskManager],
) -> None

from tenxgraph.runtime.publisher import publish_event
from tenxgraph.runtime.publisher.events import ContentType, Event, EventModel, EventType
from tenxgraph.utils import tool


@tool
def shout(query: str) -> str:
    """Uppercase the query and emit a custom event."""
    result = query.upper()
    publish_event(
        EventModel(
            event=Event.TOOL_EXECUTION,
            event_type=EventType.RESULT,
            node_name="shout",
            content=result,
            content_type=[ContentType.TEXT],
            data={"query": query, "result_length": len(result)},
        )
    )
    return result

import asyncio
from typing import Any

import httpx

from tenxgraph.core.graph import StateGraph
from tenxgraph.runtime.publisher import BasePublisher
from tenxgraph.runtime.publisher.events import EventModel


class WebhookPublisher(BasePublisher):
    """Post each event as JSON to a webhook."""

    def __init__(self, webhook_url: str, config: dict[str, Any] | None = None):
        super().__init__(config or {})
        self.webhook_url = webhook_url
        self._client: httpx.AsyncClient | None = None

    async def publish(self, event: EventModel) -> None:
        if self._is_closed:
            raise RuntimeError("Cannot publish to closed WebhookPublisher")
        if self._client is None:
            self._client = httpx.AsyncClient()
        await self._client.post(self.webhook_url, json=event.model_dump(mode="json"))

    async def close(self) -> None:
        if self._is_closed:
            return
        if self._client is not None:
            await self._client.aclose()
            self._client = None
        self._is_closed = True

    def sync_close(self) -> None:
        asyncio.run(self.close())


publisher = WebhookPublisher("https://example.com/events")
graph = StateGraph(publisher=publisher)