Custom Context Services – API Guide (using `Service` alias)¶

AetherGraph lets you attach your own runtime helpers onto context.* as first‑class services (e.g., context.trainer(), context.materials()).

Use Service (alias of the underlying BaseContextService) for all subclasses:
from aethergraph import Service  # alias of BaseContextService

Register your service after the sidecar starts, then access it anywhere via the context: context.<name>().

Minimal Flow¶

from aethergraph import start_server, Service
from aethergraph.runtime import register_context_service

class MyCache(Service):
    def __init__(self):
        super().__init__()
        self._data = {}

    def get(self, k, default=None):
        return self._data.get(k, default)

    def set(self, k, v):
        self._data[k] = v

# 1) Start sidecar
start_server(port=0)

# 2) Register a **singleton** instance for all contexts
register_context_service("cache", MyCache())

# 3) Use inside a graph/tool
async def do_work(*, context):
    cache = context.cache()        # no‑arg call ⇒ bound service instance
    cache.set("foo", 42)
    return cache.get("foo")

Why it works:

context.__getattr__("cache") resolves the registered service and binds the active NodeContext via Service.bind(...).
The returned _ServiceHandle forwards attribute access and calls. A no‑arg call returns the underlying service instance for convenience.

API Reference¶

1. Register External Services¶

register_context_service(name, service)

Register an external service for NodeContext access.

This function attaches an external service to the current service container under the specified name. If no container is installed yet, the service is stashed in a pending registry and will be attached automatically when install_services() is called.

Examples:

Register a custom database service:

register_context_service("mydb", MyDatabaseService())

Parameters:

Name	Type	Description	Default
`name`	`str`	The unique string identifier for the external service.	required
`service`	`Service`	The service instance to register.	required

Returns:

Type	Description
`None`	None

Notes

If called before install_services(), the service will be attached later.
Services are accessible via NodeContext.ext_services[name].

get_ext_context_service(name)

Retrieve an external context service by name.

This function returns the external service registered under the given name from the current service container's ext_services registry.

Examples:

Access a registered service:

mydb = get_ext_context_service("mydb")

Parameters:

Name	Type	Description	Default
`name`	`str`	The string name of the external service to retrieve.	required

Returns:

Type	Description
`Service`	The service instance registered under the given name, or None if not found.

Raises:

Type	Description
`RuntimeError`	If no services container is installed.

list_ext_context_services()

List all registered external context service names.

This function returns a list of all names for services currently registered in the ext_services registry of the current service container.

Examples:

List all available external services:

services = list_ext_context_services()
print(services)

Returns:

Type	Description
`list[str]`	A list of strings representing the names of all registered external services.
`list[str]`	Returns an empty list if no services are registered.

Raises:

Type	Description
`RuntimeError`	If no services container is installed.

2. Access External Services¶

context.<service_name>(*args)

Retrieve and bind an external context service by name. This allows accessing services as attributes on the context object.

This method overrides attribute access to dynamically resolve external services registered in the context. If a service with the requested name exists, it is retrieved and wrapped in a _ServiceHandle for ergonomic access. The returned handle allows attribute access, direct retrieval, and call forwarding if the service is callable.

Examples:

# Retrieve a database service and run a query
db = context.database()
db.query("SELECT * FROM users")

# Access a logger service and log a message
context.logger.info("Hello from node!")

# Forward arguments to a callable service
result = context.some_tool("input text")

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the service to resolve as an attribute.	required

Returns:

Name	Type	Description
`_ServiceHandle`	`Any`	A callable handle to the resolved service.

Raises:

Type	Description
`AttributeError`	If no service with the given name exists in the context.

Usage

You can access external services directly as attributes on the context object. For example, if you have registered a service named "my_service", you can use:

# Get the service instance
svc = context.my_service()

# Call the service if it's callable
result = context.my_service(arg1, arg2)

# Access service attributes
value = context.my_service.some_attribute

In your Service, you can use self.ctx to access the node context if needed. For example:

class MyService:
    ...
    def my_method(self, ...):
        context = self.ctx  # Access the NodeContext
        # Use context information as needed
        context.channel.send("Hello from MyService!")

Notes

If the service is not registered, an AttributeError is raised.
If the service is callable, calling context.service_name(args) will forward the call.
If you call context.service_name() with no arguments, you get the underlying service instance.
Attribute access (e.g., context.service_name.some_attr) is delegated to the service.

context.svc(name)

Retrieve and bind an external context service by name. This method is equivalent to context.<service_name>(). User can use either context.svc("service_name") or context.service_name() to access the service.

This method accesses a registered external service, optionally binding it to the current node context if the service supports context binding via a bind method.

Examples:

Basic usage to access a service:

db = context.svc("database")

Accessing a service that requires context binding:

logger = context.svc("logger")
logger.info("Node started.")

Parameters:

Name	Type	Description	Default
`name`	`str`	The unique string identifier of the external service to retrieve.	required

Returns:

Name	Type	Description
`Any`	`Any`	The external service instance, bound to the current context if applicable.

Raises:

Type	Description
`KeyError`	If the requested service is not registered in the external context.

Patterns¶

1) Context‑Aware Logging & Artifacts¶

from aethergraph import Service

class Reporter(Service):
    def log_metric(self, name: str, value: float):
        self.ctx().logger().info("metric", extra={"name": name, "value": value})

    async def save_text(self, text: str) -> str:
        art = await self.ctx().artifacts().save_text(text)
        return art.uri

2) Thread‑safe Mutations¶

from aethergraph import Service

class Counter(Service):
    def __init__(self):
        super().__init__()
        self._n = 0
        self.inc = self.critical()(self.inc)

    def inc(self, k: int = 1) -> int:
        self._n += k
        return self._n

3) Wrapping External Clients¶

from aethergraph import Service
import httpx

class Weather(Service):
    def __init__(self, base: str):
        super().__init__()
        self._base = base
        self._http = httpx.Client(timeout=10)

    async def close(self):
        await self.run_blocking(self._http.close)

    def get_temp_c(self, city: str) -> float:
        r = self._http.get(f"{self._base}/temp", params={"city": city})
        r.raise_for_status()
        return float(r.json()["c"])  # noqa

4) Using in Graphs/Tools¶

from aethergraph import Service
from aethergraph.runtime import register_context_service

class Greeter(Service):
    def greet(self, name: str) -> str:
        return f"Hello, {name}! (run={self.ctx().run_id})"

register_context_service("greeter", Greeter())

# Inside a node
async def hello(*, context):
    g = context.greeter()               # bound service
    return {"msg": g.greet("World")}

FAQ¶

Q: Where should I create the service instance? Usually at process boot, after start_server(...). Register exactly once (singleton). If you need per‑tenant services, implement your own map inside the service keyed by tenant.

Q: How do I access other context services from my service? Use self.ctx().<other_service>(), e.g., self.ctx().artifacts() or self.ctx().logger().

Q: Can a service be async? Yes—methods can be async and you may leverage run_blocking(...) for sync IO.

Q: How do I store configuration/keys? Provide them to your service constructor, or pull from env. For dynamic secrets, wire a secrets service and read from it inside your custom service.

Gotchas & Tips¶

Name collisions: Avoid names of built‑ins (channel, memory, artifacts, etc.).
No context at import time: Service.ctx() works after binding; do not call it in __init__.
Thread safety: Use @Service.critical for shared state.
Long‑running IO: Prefer await self.run_blocking(...) for blocking clients.
Testing: You can construct a Service and call bind(context=FakeContext(...)) directly in unit tests.

Custom Context Services – API Guide (using Service alias)¶

Minimal Flow¶

API Reference¶

1. Register External Services¶

2. Access External Services¶

Patterns¶

1) Context‑Aware Logging & Artifacts¶

2) Thread‑safe Mutations¶

3) Wrapping External Clients¶

4) Using in Graphs/Tools¶

FAQ¶

Gotchas & Tips¶

Custom Context Services – API Guide (using `Service` alias)¶