Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ build/
.mypy_cache/
.ruff_cache/
uv.lock
.idea/
.DS_Store

# === agent-memory: AI infrastructure (personal / per-machine — do not commit) ===
Expand Down
21 changes: 18 additions & 3 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -26,10 +26,10 @@ graph task calls a Python function exactly as if it were local.

```python
# app.py
from mercury_composable import AppException, platform, preload
from mercury_composable import AppException, Body, platform, preload

@preload(route="hello.python", instances=10)
def handle_event(headers: dict, body):
def handle_event(headers: dict[str, str], body: Body):
if not isinstance(body, dict) or "text" not in body:
raise AppException(400, "missing 'text'")
return {"text": str(body["text"]).upper(), "language": "python"}
Expand Down Expand Up @@ -66,7 +66,7 @@ now executes the Python function, with trace context carried end to end.
## The function contract

A handler receives the same two-part input as an engine `TypedLambdaFunction` —
`(headers: dict, body)` — and returns the reply body (or an `EventEnvelope` for full
`(headers: dict[str, str], body: Body)` — `Body` is any MsgPack value — and returns the reply body (or an `EventEnvelope` for full
control of status and reply headers). `async def` and plain `def` are both supported;
synchronous handlers run in a thread-pool executor so the event loop never blocks.

Expand All @@ -75,6 +75,8 @@ synchronous handlers run in a thread-pool executor so the event loop never block
exception handler or the graph's `error.*` contract.
- `get_trace()` exposes `trace_id` / `trace_path` / `cid`; `annotate_trace(k, v)` sends an
annotation back on the reply envelope.
- Outside a hosted function (batch jobs, tests), `trace_context(trace_id, trace_path)`
establishes the context your `PostOffice` calls inherit — the node `runWithTrace` twin.
- Functions must be stateless; anything you must keep belongs to the caller's flow model
or state machine.

Expand Down Expand Up @@ -125,6 +127,19 @@ orchestration** — those live in the engines. It provides functions plus the mi
foundation utilities, keeping Python fast to prototype with while the composable core
guarantees the architecture.

## Development

```bash
uv venv .venv && uv pip install -e '.[dev]' # environment (uv-managed python)
.venv/bin/pytest -q # tests
uvx ruff check . # lint (config in pyproject.toml)
uvx basedpyright # type check (config in pyproject.toml)
```

PyCharm: use interpreter type **uv** pointing at the project `.venv`, and set
*Settings → Tools → Python Integrated Tools → Package requirements file* to
`pyproject.toml` so the requirements inspection reads `[project.dependencies]`.

## License

Apache 2.0 — see [LICENSE.txt](LICENSE.txt).
20 changes: 16 additions & 4 deletions examples/demo_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,14 +10,26 @@
target: 'http://127.0.0.1:8086/api/event'
"""

from mercury_composable import AppException, annotate_trace, get_logger, platform, preload
from mercury_composable import (
AppException,
Body,
annotate_trace,
get_logger,
platform,
preload,
)

log = get_logger(__name__)


@preload(route="hello.python", instances=10)
def handle_event(headers: dict, body):
"""Uppercase transform - the polyglot hello world."""
def handle_event(_headers: dict[str, str], body: Body):
"""Uppercase transform - the polyglot hello world.

The (headers, body) two-part signature is the function contract (the
TypedLambdaFunction mirror) - a handler that does not need headers keeps
the parameter, underscore-prefixed per Python convention.
"""
if not isinstance(body, dict) or "text" not in body:
raise AppException(400, "missing 'text'")
annotate_trace("language", "python")
Expand All @@ -26,7 +38,7 @@ def handle_event(headers: dict, body):


@preload(route="hello.declarative", instances=10)
async def declarative_echo(headers: dict, body):
async def declarative_echo(headers: dict[str, str], body: Body):
"""Echo for the composable-example declarative Event-over-HTTP demo."""
return {"body": body, "headers": headers, "language": "python"}

Expand Down
23 changes: 22 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -27,14 +27,35 @@ Homepage = "https://github.com/Accenture/mercury-python"
Documentation = "https://accenture.github.io/mercury-composable"

[project.optional-dependencies]
dev = ["pytest>=8", "pytest-asyncio>=0.23"]
dev = ["pytest>=8", "pytest-asyncio>=0.23", "ruff>=0.9"]

[project.scripts]
mercury-serve = "mercury_composable.cli:main"

[tool.hatch.build.targets.wheel]
packages = ["src/mercury_composable"]

[tool.ruff]
line-length = 100
target-version = "py310"
# the agent-skills layer is tool-managed by agent-memory (overwritten on upgrade);
# style fixes for it belong upstream, not here
extend-exclude = ["agent-skills"]

[tool.ruff.lint]
extend-select = ["I"]

[tool.basedpyright]
# standard mode plus the contract-relevant strictness: every function parameter
# is annotated, unit tests included (the wire contract is typed - Body,
# dict[str, str]); the reportUnknown* warning family of the editor's
# "recommended" mode is deliberately not chased. agent-skills is tool-managed
# by agent-memory and excluded.
typeCheckingMode = "standard"
reportMissingParameterType = "error"
include = ["src", "examples", "tests"]
exclude = ["agent-skills", ".venv"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
32 changes: 25 additions & 7 deletions src/mercury_composable/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,18 +11,36 @@

from .client import PostOffice
from .config import AppConfig, app_config, load_config
from .envelope import EventEnvelope, iso_utc
from .envelope import Body, EventEnvelope, iso_utc
from .exceptions import AppException, CompactFormatError
from .log import get_logger
from .registry import FunctionRegistry, default_registry, preload
from .registry import FunctionRegistry, Handler, default_registry, preload
from .server import EventApiServer, Platform, platform
from .trace import TraceInfo, annotate_trace, get_trace
from .trace import TraceInfo, annotate_trace, get_trace, trace_context

__version__ = "0.1.0"

__all__ = [
"AppConfig", "AppException", "CompactFormatError", "EventApiServer",
"EventEnvelope", "FunctionRegistry", "Platform", "PostOffice", "TraceInfo",
"annotate_trace", "app_config", "default_registry", "get_logger",
"get_trace", "iso_utc", "load_config", "platform", "preload", "__version__",
"AppConfig",
"AppException",
"Body",
"CompactFormatError",
"EventApiServer",
"EventEnvelope",
"FunctionRegistry",
"Handler",
"Platform",
"PostOffice",
"TraceInfo",
"__version__",
"annotate_trace",
"app_config",
"default_registry",
"get_logger",
"get_trace",
"iso_utc",
"load_config",
"platform",
"preload",
"trace_context",
]
11 changes: 8 additions & 3 deletions src/mercury_composable/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,15 @@ def main() -> int:
help="Configuration file (default: resources/application.yml|properties)")
# -Dkey=value runtime overrides are consumed by AppConfig from sys.argv
args, _unknown = parser.parse_known_args()
# argparse Namespace attributes are untyped - pin the types at the boundary
app_arg: str = args.app
port_arg: int | None = args.port
host_arg: str = args.host
config_arg: str | None = args.config

from .config import DEFAULT_CANDIDATES, load_config
app_path = os.path.abspath(args.app)
config_path = args.config
app_path = os.path.abspath(app_arg)
config_path: str | None = config_arg
if config_path is None and not any(os.path.isfile(c) for c in DEFAULT_CANDIDATES):
# fall back to a resources folder next to the application file
app_dir = os.path.dirname(app_path)
Expand Down Expand Up @@ -56,7 +61,7 @@ def main() -> int:
print("No functions registered - use @preload(route=..., instances=...)",
file=sys.stderr)
return 1
platform.run(port=args.port, host=args.host)
platform.run(port=port_arg, host=host_arg)
return 0


Expand Down
81 changes: 44 additions & 37 deletions src/mercury_composable/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
from __future__ import annotations

import re
from typing import Any, Dict, Optional
from typing import Any

import aiohttp

Expand All @@ -27,32 +27,52 @@
_W3C_SPAN_ID = re.compile(r"^[0-9a-f]{16}$")


def _build_event(route: str, body: Any, headers: dict[str, str] | None,
from_route: str | None, cid: str | None) -> EventEnvelope:
"""Build the outbound envelope, inheriting the current trace context."""
event = EventEnvelope(to=route, body=body, headers=headers or {})
if from_route:
event.set_from(from_route)
info = get_trace()
if info and info.trace_id:
event.set_trace(info.trace_id, info.trace_path or route)
effective_cid = cid or (info.cid if info else None)
if effective_cid:
event.set_correlation_id(effective_cid)
return event


class PostOffice:
"""Event-over-HTTP client for calling functions on peer applications."""

def __init__(self, endpoint: Optional[str] = None,
security_headers: Optional[Dict[str, str]] = None):
def __init__(self, endpoint: str | None = None,
security_headers: dict[str, str] | None = None):
self.endpoint = endpoint
self.security_headers = dict(security_headers or {})
self._session: Optional[aiohttp.ClientSession] = None
self._session: aiohttp.ClientSession | None = None

async def _get_session(self) -> aiohttp.ClientSession:
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession()
return self._session
def _get_session(self) -> aiohttp.ClientSession:
# called from the running event loop only (inside request/send)
session = self._session
if session is None or session.closed:
session = aiohttp.ClientSession()
self._session = session
return session

async def close(self) -> None:
if self._session is not None and not self._session.closed:
await self._session.close()

async def __aenter__(self) -> "PostOffice":
# PYI034 wants '-> Self', which needs python >= 3.11; switch when the
# floor moves past 3.10
async def __aenter__(self) -> PostOffice: # noqa: PYI034
return self

async def __aexit__(self, *_exc) -> None:
async def __aexit__(self, *_exc: object) -> None:
await self.close()

def _http_headers(self, timeout_ms: int, is_async: bool,
event: EventEnvelope) -> Dict[str, str]:
event: EventEnvelope) -> dict[str, str]:
headers = {
"content-type": "application/octet-stream",
"accept": "*/*",
Expand All @@ -71,28 +91,15 @@ def _http_headers(self, timeout_ms: int, is_async: bool,
headers["traceparent"] = f"00-{event.trace_id}-{event.span_id}-01"
return headers

def _build_event(self, route: str, body: Any, headers: Optional[Dict[str, str]],
from_route: Optional[str], cid: Optional[str]) -> EventEnvelope:
event = EventEnvelope(to=route, body=body, headers=headers or {})
if from_route:
event.set_from(from_route)
info = get_trace()
if info and info.trace_id:
event.set_trace(info.trace_id, info.trace_path or route)
effective_cid = cid or (info.cid if info else None)
if effective_cid:
event.set_correlation_id(effective_cid)
return event

async def _call(self, route: str, body: Any, headers: Optional[Dict[str, str]],
timeout_ms: int, endpoint: Optional[str], is_async: bool,
from_route: Optional[str], cid: Optional[str]) -> EventEnvelope:
async def _call(self, route: str, body: Any, headers: dict[str, str] | None,
timeout_ms: int, endpoint: str | None, is_async: bool,
from_route: str | None, cid: str | None) -> EventEnvelope:
url = endpoint or self.endpoint
if not url:
raise ValueError("Missing event endpoint - "
"e.g. PostOffice(endpoint='http://peer:8085/api/event')")
event = self._build_event(route, body, headers, from_route, cid)
session = await self._get_session()
event = _build_event(route, body, headers, from_route, cid)
session = self._get_session()
# +100 ms cushion so the HTTP client does not time out before the target
client_timeout = aiohttp.ClientTimeout(total=(max(100, timeout_ms) + 100) / 1000)
async with session.post(url, data=event.to_bytes(),
Expand All @@ -106,21 +113,21 @@ async def _call(self, route: str, body: Any, headers: Optional[Dict[str, str]],
f"Invalid event-over-http response - {e}") from e

async def request(self, route: str, body: Any = None, *,
headers: Optional[Dict[str, str]] = None,
headers: dict[str, str] | None = None,
timeout_ms: int = 30000,
endpoint: Optional[str] = None,
from_route: Optional[str] = None,
cid: Optional[str] = None) -> EventEnvelope:
endpoint: str | None = None,
from_route: str | None = None,
cid: str | None = None) -> EventEnvelope:
"""RPC call: returns the target function's reply envelope."""
return await self._call(route, body, headers, timeout_ms, endpoint,
False, from_route, cid)

async def send(self, route: str, body: Any = None, *,
headers: Optional[Dict[str, str]] = None,
headers: dict[str, str] | None = None,
timeout_ms: int = 30000,
endpoint: Optional[str] = None,
from_route: Optional[str] = None,
cid: Optional[str] = None) -> EventEnvelope:
endpoint: str | None = None,
from_route: str | None = None,
cid: str | None = None) -> EventEnvelope:
"""Drop-n-forget: returns the peer's 202 delivery acknowledgement envelope."""
return await self._call(route, body, headers, timeout_ms, endpoint,
True, from_route, cid)
Loading
Loading