mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: add agent-framework-hosting-invocations channel (#5640)
* feat(hosting-invocations): add Invocations channel package New ``agent-framework-hosting-invocations`` package implementing the "Invocations" HTTP channel for the Hosting framework -- a lightweight JSON-over-HTTP shape (``POST /invocations``) for callers that want a single request/response without committing to the full OpenAI Responses envelope. Mounts onto an ``AgentFrameworkHost`` like any other channel. Surface (re-exported from ``agent_framework_hosting_invocations``): - ``InvocationsChannel`` -- concrete ``Channel`` implementation. Owns the Starlette route, parses inbound JSON into a ``ChannelRequest`` (``input`` / ``session`` / ``metadata`` / ``options``), runs the optional ``ChannelRunHook``, calls back into the ``ChannelContext`` to invoke the agent target, and returns a flat JSON envelope (or an SSE stream when ``stream=true``). - 8 unit tests covering route wiring, isolation-key passthrough, hook composition, sync vs streaming paths, and ack-only behaviour for non-originating ``DeliveryReport``s. Registers the package in ``python/pyproject.toml`` ``[tool.uv.sources]`` and adds the matching pyright ``executionEnvironments`` entry. Independent of PR-3 (Responses); both depend only on PR-2 (Hosting core). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * review: address PR-4 round 2 feedback - expand `_stream` docstring to call out the HTTP-200 + `event: error` SSE contract (status committed before generator runs; hard failures surface as the first SSE frame, not an HTTP code) - split chunked text on full-line terminators via `splitlines()` so embedded `\r` / `\r\n` no longer leak into `data:` framing on the wire, breaking EventSource consumers - on `get_final_response()` failure, emit `event: error` instead of silently swallowing — finalize is what triggers history-provider persistence on the agent side, so a 5xx / disk-full / context-provider error must reach the client - add tests covering `stream_transform_hook` (rewrite, drop, async), CRLF-in-chunk framing, and the finalize-error → no-`[DONE]` contract Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * docs(hosting-invocations): rename stale ChatMessage docstring reference to Message Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix(hosting-invocations): adapt to hosted run result wrapper Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * feat(hosting-invocations): add response hooks Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Eduard van Valkenburg
committed by
GitHub
Unverified
GitHub
parent
d75f55ee2c
commit
cb1d4a6ee5
@@ -0,0 +1,21 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) Microsoft Corporation.
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE
|
||||
@@ -0,0 +1,30 @@
|
||||
# agent-framework-hosting-invocations
|
||||
|
||||
Minimal `POST /invoke` channel for [agent-framework-hosting](../hosting). Useful
|
||||
for smoke-testing, durable-task drivers, and bespoke clients that don't speak
|
||||
the OpenAI Responses protocol.
|
||||
|
||||
## Wire shape
|
||||
|
||||
```
|
||||
POST /invocations/invoke
|
||||
{
|
||||
"message": "hello",
|
||||
"session_id": "user-42",
|
||||
"stream": false
|
||||
}
|
||||
```
|
||||
|
||||
Non-streaming response: `{"response": "...", "session_id": "..."}`.
|
||||
Streaming response: `text/event-stream` of `data:` lines, terminated by
|
||||
`data: [DONE]`.
|
||||
|
||||
## Usage
|
||||
|
||||
```python
|
||||
from agent_framework_hosting import AgentFrameworkHost
|
||||
from agent_framework_hosting_invocations import InvocationsChannel
|
||||
|
||||
host = AgentFrameworkHost(target=my_agent, channels=[InvocationsChannel()])
|
||||
host.serve()
|
||||
```
|
||||
@@ -0,0 +1,7 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Minimal ``POST /invoke`` channel for :mod:`agent_framework_hosting`."""
|
||||
|
||||
from ._channel import InvocationsChannel
|
||||
|
||||
__all__ = ["InvocationsChannel"]
|
||||
@@ -0,0 +1,219 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Minimal ``POST /invoke`` channel.
|
||||
|
||||
Inspired by ``agent-framework-foundry-hosting``'s ``InvocationsHostServer``.
|
||||
A framework-agnostic surface for callers that just want to send a message and
|
||||
get an answer back — no OpenAI-style envelope, no Responses item lattice.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import AsyncIterator, Awaitable
|
||||
from typing import Any, cast
|
||||
|
||||
from agent_framework_hosting import (
|
||||
ChannelContext,
|
||||
ChannelContribution,
|
||||
ChannelRequest,
|
||||
ChannelResponseContext,
|
||||
ChannelResponseHook,
|
||||
ChannelRunHook,
|
||||
ChannelSession,
|
||||
ChannelStreamTransformHook,
|
||||
HostedRunResult,
|
||||
apply_response_hook,
|
||||
apply_run_hook,
|
||||
logger,
|
||||
)
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import JSONResponse, Response, StreamingResponse
|
||||
from starlette.routing import Route
|
||||
|
||||
|
||||
class InvocationsChannel:
|
||||
"""Minimal ``POST /invoke`` surface.
|
||||
|
||||
A run hook can rewrite the channel request (e.g. inject a session, add
|
||||
options) before the host invokes the agent. A stream-transform hook can
|
||||
rewrite or drop ``AgentResponseUpdate`` chunks before they hit the wire.
|
||||
"""
|
||||
|
||||
name = "invocations"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
path: str = "/invocations",
|
||||
run_hook: ChannelRunHook | None = None,
|
||||
response_hook: ChannelResponseHook | None = None,
|
||||
stream_transform_hook: ChannelStreamTransformHook | None = None,
|
||||
) -> None:
|
||||
"""Configure the invocations endpoint.
|
||||
|
||||
``path`` is the mount root the host prefixes when registering this
|
||||
channel's routes (the actual handler is ``POST {path}/invoke``).
|
||||
``run_hook`` may rewrite the :class:`ChannelRequest` before the host
|
||||
invokes the target — typically to attach session metadata or
|
||||
translate the wire payload into ``Message`` instances.
|
||||
``response_hook`` may rewrite the :class:`HostedRunResult` before
|
||||
the channel serializes it to JSON for the originating caller.
|
||||
``stream_transform_hook`` lets callers map or drop individual
|
||||
``AgentResponseUpdate`` chunks while streaming.
|
||||
"""
|
||||
self.path = path
|
||||
self._hook = run_hook
|
||||
self.response_hook = response_hook
|
||||
self._stream_transform_hook = stream_transform_hook
|
||||
self._ctx: ChannelContext | None = None
|
||||
|
||||
def contribute(self, context: ChannelContext) -> ChannelContribution:
|
||||
"""Capture the host-supplied context and register ``POST /invoke``."""
|
||||
self._ctx = context
|
||||
return ChannelContribution(routes=[Route("/invoke", self._handle, methods=["POST"])])
|
||||
|
||||
async def _handle(self, request: Request) -> Response:
|
||||
"""Handle a single ``POST /invoke`` call.
|
||||
|
||||
Validates the JSON body shape, builds a :class:`ChannelRequest`
|
||||
(optionally with a ``ChannelSession`` keyed by ``session_id``),
|
||||
runs the configured ``run_hook``, and either streams SSE chunks
|
||||
when ``stream`` is true or returns a single JSON ``{response,
|
||||
session_id}`` envelope.
|
||||
"""
|
||||
if self._ctx is None: # pragma: no cover - guarded by Channel lifecycle
|
||||
return JSONResponse({"error": "channel not initialized"}, status_code=500)
|
||||
try:
|
||||
body: Any = await request.json()
|
||||
except Exception:
|
||||
return JSONResponse({"error": "invalid json"}, status_code=400)
|
||||
|
||||
if not isinstance(body, dict):
|
||||
return JSONResponse({"error": "request body must be an object"}, status_code=422)
|
||||
body_map: dict[str, Any] = cast("dict[str, Any]", body)
|
||||
|
||||
message = body_map.get("message")
|
||||
if not isinstance(message, str) or not message:
|
||||
return JSONResponse({"error": "missing or empty 'message'"}, status_code=422)
|
||||
|
||||
session_id = body_map.get("session_id")
|
||||
if session_id is not None and not isinstance(session_id, str):
|
||||
return JSONResponse({"error": "'session_id' must be a string"}, status_code=422)
|
||||
|
||||
session = ChannelSession(isolation_key=f"invocations:{session_id}") if session_id else None
|
||||
|
||||
attributes: dict[str, Any] = {}
|
||||
if session_id:
|
||||
attributes["session_id"] = session_id
|
||||
|
||||
channel_request = ChannelRequest(
|
||||
channel=self.name,
|
||||
operation="invoke",
|
||||
input=message,
|
||||
session=session,
|
||||
stream=bool(body_map.get("stream")),
|
||||
attributes=attributes,
|
||||
)
|
||||
|
||||
if self._hook is not None:
|
||||
channel_request = await apply_run_hook(
|
||||
self._hook,
|
||||
channel_request,
|
||||
target=self._ctx.target,
|
||||
protocol_request=body_map,
|
||||
)
|
||||
|
||||
if channel_request.stream:
|
||||
return StreamingResponse(
|
||||
self._stream(channel_request),
|
||||
media_type="text/event-stream",
|
||||
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
|
||||
)
|
||||
|
||||
result = await self._ctx.run(channel_request)
|
||||
result = await self._apply_response_hook(result, channel_request)
|
||||
return JSONResponse({"response": result.result.text, "session_id": session_id})
|
||||
|
||||
async def _apply_response_hook(
|
||||
self,
|
||||
result: HostedRunResult[Any],
|
||||
request: ChannelRequest,
|
||||
) -> HostedRunResult[Any]:
|
||||
"""Apply the channel-level response hook for an originating reply."""
|
||||
if self.response_hook is None:
|
||||
return result
|
||||
context = ChannelResponseContext(
|
||||
request=request,
|
||||
channel_name=self.name,
|
||||
destination_identity=None,
|
||||
originating=True,
|
||||
is_echo=False,
|
||||
)
|
||||
return await apply_response_hook(self.response_hook, result, context=context)
|
||||
|
||||
async def _stream(self, request: ChannelRequest) -> AsyncIterator[str]:
|
||||
r"""Yield bare ``data:`` SSE lines for each text chunk + a final ``[DONE]``.
|
||||
|
||||
SSE protocol notes:
|
||||
|
||||
* The HTTP status is committed when ASGI sends headers, before the
|
||||
generator runs. Emitting a stream-opening 200 + ``text/event-stream``
|
||||
and signalling errors via ``event: error`` SSE frames is the
|
||||
conventional contract — ``EventSource`` and OpenAI-style SSE
|
||||
consumers treat ``event: error`` as a terminal error condition.
|
||||
Hard run-acquisition failures (e.g. target rejected) therefore
|
||||
surface as the first frame, not as an HTTP error code.
|
||||
* The SSE spec treats ``\r``, ``\n``, and ``\r\n`` as line
|
||||
terminators. Per-chunk text is split on all three so embedded
|
||||
carriage returns don't corrupt ``data:`` framing on the wire.
|
||||
"""
|
||||
if self._ctx is None: # pragma: no cover - guarded by Channel lifecycle
|
||||
yield "event: error\ndata: channel not initialized\n\n"
|
||||
return
|
||||
try:
|
||||
stream = self._ctx.run_stream(request)
|
||||
async for update in stream:
|
||||
if self._stream_transform_hook is not None:
|
||||
transformed = self._stream_transform_hook(update)
|
||||
if isinstance(transformed, Awaitable):
|
||||
transformed = await transformed
|
||||
if transformed is None:
|
||||
continue
|
||||
update = transformed
|
||||
chunk = getattr(update, "text", None)
|
||||
if chunk:
|
||||
# Each text chunk is its own SSE event so curl-friendly
|
||||
# consumers can read it directly. Newlines inside the
|
||||
# chunk are escaped per SSE spec by emitting one
|
||||
# ``data:`` line per source line. ``splitlines()`` is
|
||||
# used over ``split('\n')`` so embedded ``\r`` /
|
||||
# ``\r\n`` don't bleed into the framing.
|
||||
for line in str(chunk).splitlines() or [""]:
|
||||
yield f"data: {line}\n"
|
||||
yield "\n"
|
||||
try:
|
||||
# Finalize so context-provider / history hooks on the agent
|
||||
# still run even though we are emitting our own SSE.
|
||||
# If finalization fails, the agent's persistence side
|
||||
# effects (history-provider write, context-provider hooks)
|
||||
# are unreliable — surface that to the client as an
|
||||
# ``event: error`` frame so it isn't a silent drop.
|
||||
await stream.get_final_response()
|
||||
except Exception as finalize_exc:
|
||||
logger.exception("Invocations stream finalize failed")
|
||||
yield "event: error\n"
|
||||
for line in f"finalize failed: {finalize_exc!s}".splitlines() or [""]:
|
||||
yield f"data: {line}\n"
|
||||
yield "\n"
|
||||
return
|
||||
except Exception as exc:
|
||||
logger.exception("Invocations stream consumption failed")
|
||||
yield "event: error\n"
|
||||
for line in str(exc).splitlines() or [""]:
|
||||
yield f"data: {line}\n"
|
||||
yield "\n"
|
||||
return
|
||||
yield "data: [DONE]\n\n"
|
||||
|
||||
|
||||
__all__ = ["InvocationsChannel"]
|
||||
@@ -0,0 +1,97 @@
|
||||
[project]
|
||||
name = "agent-framework-hosting-invocations"
|
||||
description = "Minimal POST /invoke channel for agent-framework-hosting."
|
||||
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.10"
|
||||
version = "1.0.0a260424"
|
||||
license-files = ["LICENSE"]
|
||||
urls.homepage = "https://aka.ms/agent-framework"
|
||||
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
|
||||
urls.release_notes = "https://github.com/microsoft/agent-framework/releases?q=tag%3Apython-1&expanded=true"
|
||||
urls.issues = "https://github.com/microsoft/agent-framework/issues"
|
||||
classifiers = [
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Development Status :: 3 - Alpha",
|
||||
"Intended Audience :: Developers",
|
||||
"Programming Language :: Python :: 3",
|
||||
"Programming Language :: Python :: 3.10",
|
||||
"Programming Language :: Python :: 3.11",
|
||||
"Programming Language :: Python :: 3.12",
|
||||
"Programming Language :: Python :: 3.13",
|
||||
"Programming Language :: Python :: 3.14",
|
||||
"Typing :: Typed",
|
||||
]
|
||||
dependencies = [
|
||||
"agent-framework-core>=1.2.0,<2",
|
||||
"agent-framework-hosting==1.0.0a260424",
|
||||
]
|
||||
|
||||
[tool.uv]
|
||||
prerelease = "if-necessary-or-explicit"
|
||||
environments = [
|
||||
"sys_platform == 'darwin'",
|
||||
"sys_platform == 'linux'",
|
||||
"sys_platform == 'win32'"
|
||||
]
|
||||
|
||||
[tool.uv-dynamic-versioning]
|
||||
fallback-version = "0.0.0"
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = 'tests'
|
||||
addopts = "-ra -q -r fEX"
|
||||
asyncio_mode = "auto"
|
||||
asyncio_default_fixture_loop_scope = "function"
|
||||
filterwarnings = []
|
||||
timeout = 120
|
||||
markers = [
|
||||
"integration: marks tests as integration tests that require external services",
|
||||
]
|
||||
|
||||
[tool.ruff]
|
||||
extend = "../../pyproject.toml"
|
||||
|
||||
[tool.coverage.run]
|
||||
omit = [
|
||||
"**/__init__.py"
|
||||
]
|
||||
|
||||
[tool.pyright]
|
||||
extends = "../../pyproject.toml"
|
||||
include = ["agent_framework_hosting_invocations"]
|
||||
exclude = ['tests']
|
||||
|
||||
[tool.mypy]
|
||||
plugins = ['pydantic.mypy']
|
||||
strict = true
|
||||
python_version = "3.10"
|
||||
ignore_missing_imports = true
|
||||
disallow_untyped_defs = true
|
||||
no_implicit_optional = true
|
||||
check_untyped_defs = true
|
||||
warn_return_any = true
|
||||
show_error_codes = true
|
||||
warn_unused_ignores = false
|
||||
disallow_incomplete_defs = true
|
||||
disallow_untyped_decorators = true
|
||||
|
||||
[tool.bandit]
|
||||
targets = ["agent_framework_hosting_invocations"]
|
||||
exclude_dirs = ["tests"]
|
||||
|
||||
[tool.poe]
|
||||
executor.type = "uv"
|
||||
include = "../../shared_tasks.toml"
|
||||
|
||||
[tool.poe.tasks.mypy]
|
||||
help = "Run MyPy for this package."
|
||||
cmd = "mypy --config-file $POE_ROOT/pyproject.toml agent_framework_hosting_invocations"
|
||||
|
||||
[tool.poe.tasks.test]
|
||||
help = "Run the default unit test suite for this package."
|
||||
cmd = 'pytest -m "not integration" --cov=agent_framework_hosting_invocations --cov-report=term-missing:skip-covered tests'
|
||||
|
||||
[build-system]
|
||||
requires = ["flit-core >= 3.11,<4.0"]
|
||||
build-backend = "flit_core.buildapi"
|
||||
@@ -0,0 +1,256 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""End-to-end tests for :class:`InvocationsChannel`."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import AsyncIterator
|
||||
from dataclasses import dataclass, replace
|
||||
from typing import Any
|
||||
|
||||
from agent_framework_hosting import AgentFrameworkHost, ChannelRequest, HostedRunResult
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from agent_framework_hosting_invocations import InvocationsChannel
|
||||
|
||||
|
||||
@dataclass
|
||||
class _FakeAgentResponse:
|
||||
text: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class _FakeUpdate:
|
||||
text: str
|
||||
|
||||
|
||||
class _FakeStream:
|
||||
def __init__(self, chunks: list[str]) -> None:
|
||||
self._chunks = chunks
|
||||
self._final = _FakeAgentResponse(text="".join(chunks))
|
||||
|
||||
def __aiter__(self) -> AsyncIterator[_FakeUpdate]:
|
||||
async def _gen() -> AsyncIterator[_FakeUpdate]:
|
||||
for c in self._chunks:
|
||||
yield _FakeUpdate(c)
|
||||
|
||||
return _gen()
|
||||
|
||||
async def get_final_response(self) -> _FakeAgentResponse:
|
||||
return self._final
|
||||
|
||||
|
||||
class _FakeAgent:
|
||||
def __init__(self, reply: str = "hi", chunks: list[str] | None = None) -> None:
|
||||
self._reply = reply
|
||||
self._chunks = chunks or [reply]
|
||||
self.calls: list[dict[str, Any]] = []
|
||||
|
||||
def create_session(self, *, session_id: str | None = None) -> Any:
|
||||
return {"session_id": session_id}
|
||||
|
||||
def run(self, messages: Any = None, *, stream: bool = False, **kwargs: Any) -> Any:
|
||||
self.calls.append({"messages": messages, "stream": stream, "kwargs": kwargs})
|
||||
if stream:
|
||||
return _FakeStream(self._chunks)
|
||||
|
||||
async def _coro() -> _FakeAgentResponse:
|
||||
return _FakeAgentResponse(text=self._reply)
|
||||
|
||||
return _coro()
|
||||
|
||||
|
||||
def _make_client(agent: _FakeAgent | None = None) -> tuple[TestClient, _FakeAgent]:
|
||||
agent = agent or _FakeAgent()
|
||||
host = AgentFrameworkHost(target=agent, channels=[InvocationsChannel()])
|
||||
return TestClient(host.app), agent
|
||||
|
||||
|
||||
class TestInvocations:
|
||||
def test_post_invoke_returns_response(self) -> None:
|
||||
client, _agent = _make_client(_FakeAgent(reply="pong"))
|
||||
with client:
|
||||
r = client.post("/invocations/invoke", json={"message": "ping"})
|
||||
assert r.status_code == 200
|
||||
assert r.json() == {"response": "pong", "session_id": None}
|
||||
|
||||
def test_session_id_propagates_to_target(self) -> None:
|
||||
client, agent = _make_client()
|
||||
with client:
|
||||
r = client.post("/invocations/invoke", json={"message": "x", "session_id": "s1"})
|
||||
assert r.status_code == 200
|
||||
assert r.json()["session_id"] == "s1"
|
||||
sess = agent.calls[0]["kwargs"].get("session")
|
||||
# Host converts ChannelSession.isolation_key -> AgentSession via
|
||||
# target.create_session(session_id=...). Our fake stashes that here.
|
||||
assert sess is not None
|
||||
assert sess["session_id"] == "invocations:s1"
|
||||
|
||||
def test_invalid_json_returns_400(self) -> None:
|
||||
client, _ = _make_client()
|
||||
with client:
|
||||
r = client.post(
|
||||
"/invocations/invoke",
|
||||
content=b"{not json",
|
||||
headers={"content-type": "application/json"},
|
||||
)
|
||||
assert r.status_code == 400
|
||||
|
||||
def test_empty_message_returns_422(self) -> None:
|
||||
client, _ = _make_client()
|
||||
with client:
|
||||
r = client.post("/invocations/invoke", json={"message": ""})
|
||||
assert r.status_code == 422
|
||||
|
||||
def test_non_string_session_id_returns_422(self) -> None:
|
||||
client, _ = _make_client()
|
||||
with client:
|
||||
r = client.post("/invocations/invoke", json={"message": "x", "session_id": 1})
|
||||
assert r.status_code == 422
|
||||
|
||||
def test_non_object_body_returns_422(self) -> None:
|
||||
client, _ = _make_client()
|
||||
with client:
|
||||
r = client.post("/invocations/invoke", json=[])
|
||||
assert r.status_code == 422
|
||||
|
||||
def test_streaming_emits_data_lines_and_done(self) -> None:
|
||||
agent = _FakeAgent(chunks=["hel", "lo"])
|
||||
host = AgentFrameworkHost(target=agent, channels=[InvocationsChannel()])
|
||||
with TestClient(host.app) as client:
|
||||
r = client.post("/invocations/invoke", json={"message": "x", "stream": True})
|
||||
assert r.status_code == 200
|
||||
body = r.text
|
||||
assert "data: hel" in body
|
||||
assert "data: lo" in body
|
||||
assert body.rstrip().endswith("data: [DONE]")
|
||||
|
||||
def test_run_hook_can_rewrite_request(self) -> None:
|
||||
captured: list[ChannelRequest] = []
|
||||
|
||||
async def hook(req: ChannelRequest, **_: Any) -> ChannelRequest:
|
||||
captured.append(req)
|
||||
# Force stream off even if requested.
|
||||
return replace(req, stream=False)
|
||||
|
||||
agent = _FakeAgent(reply="ok")
|
||||
host = AgentFrameworkHost(target=agent, channels=[InvocationsChannel(run_hook=hook)])
|
||||
with TestClient(host.app) as client:
|
||||
r = client.post("/invocations/invoke", json={"message": "x", "stream": True})
|
||||
assert r.status_code == 200
|
||||
# Even though caller asked for stream=True, hook flipped it off — so
|
||||
# we get JSON back, not SSE.
|
||||
assert r.headers["content-type"].startswith("application/json")
|
||||
assert captured and captured[0].channel == "invocations"
|
||||
|
||||
def test_response_hook_can_rewrite_originating_reply(self) -> None:
|
||||
contexts: list[Any] = []
|
||||
|
||||
def hook(result: HostedRunResult, **kwargs: Any) -> HostedRunResult:
|
||||
contexts.append(kwargs["context"])
|
||||
return HostedRunResult(_FakeAgentResponse(text=f"hooked:{result.result.text}"), session=result.session)
|
||||
|
||||
agent = _FakeAgent(reply="pong")
|
||||
host = AgentFrameworkHost(target=agent, channels=[InvocationsChannel(response_hook=hook)])
|
||||
|
||||
with TestClient(host.app) as client:
|
||||
r = client.post("/invocations/invoke", json={"message": "ping"})
|
||||
|
||||
assert r.status_code == 200
|
||||
assert r.json() == {"response": "hooked:pong", "session_id": None}
|
||||
assert contexts
|
||||
assert contexts[0].channel_name == "invocations"
|
||||
assert contexts[0].originating is True
|
||||
assert contexts[0].destination_identity is None
|
||||
|
||||
def test_stream_transform_hook_can_rewrite_chunks(self) -> None:
|
||||
agent = _FakeAgent(chunks=["foo", "bar"])
|
||||
|
||||
def transform(update: Any) -> Any:
|
||||
return _FakeUpdate(text=update.text.upper())
|
||||
|
||||
host = AgentFrameworkHost(
|
||||
target=agent,
|
||||
channels=[InvocationsChannel(stream_transform_hook=transform)],
|
||||
)
|
||||
with TestClient(host.app) as client:
|
||||
r = client.post("/invocations/invoke", json={"message": "x", "stream": True})
|
||||
assert r.status_code == 200
|
||||
body = r.text
|
||||
assert "data: FOO" in body
|
||||
assert "data: BAR" in body
|
||||
assert "data: foo" not in body
|
||||
|
||||
def test_stream_transform_hook_can_drop_chunks(self) -> None:
|
||||
agent = _FakeAgent(chunks=["keep", "drop", "keep2"])
|
||||
|
||||
def transform(update: Any) -> Any:
|
||||
return None if update.text == "drop" else update
|
||||
|
||||
host = AgentFrameworkHost(
|
||||
target=agent,
|
||||
channels=[InvocationsChannel(stream_transform_hook=transform)],
|
||||
)
|
||||
with TestClient(host.app) as client:
|
||||
r = client.post("/invocations/invoke", json={"message": "x", "stream": True})
|
||||
assert r.status_code == 200
|
||||
body = r.text
|
||||
assert "data: keep" in body
|
||||
assert "data: keep2" in body
|
||||
assert "data: drop" not in body
|
||||
|
||||
def test_stream_transform_hook_supports_async(self) -> None:
|
||||
agent = _FakeAgent(chunks=["aa"])
|
||||
|
||||
async def transform(update: Any) -> Any:
|
||||
return _FakeUpdate(text=update.text + "!")
|
||||
|
||||
host = AgentFrameworkHost(
|
||||
target=agent,
|
||||
channels=[InvocationsChannel(stream_transform_hook=transform)],
|
||||
)
|
||||
with TestClient(host.app) as client:
|
||||
r = client.post("/invocations/invoke", json={"message": "x", "stream": True})
|
||||
assert r.status_code == 200
|
||||
assert "data: aa!" in r.text
|
||||
|
||||
def test_streaming_chunk_with_crlf_splits_into_separate_data_lines(self) -> None:
|
||||
# Per SSE spec, ``\r``, ``\n`` and ``\r\n`` are all line terminators;
|
||||
# a chunk like ``"line1\r\nline2"`` must produce two ``data:`` lines,
|
||||
# not one ``data:`` line containing an embedded ``\r``.
|
||||
agent = _FakeAgent(chunks=["line1\r\nline2"])
|
||||
host = AgentFrameworkHost(target=agent, channels=[InvocationsChannel()])
|
||||
with TestClient(host.app) as client:
|
||||
r = client.post("/invocations/invoke", json={"message": "x", "stream": True})
|
||||
assert r.status_code == 200
|
||||
body = r.text
|
||||
assert "data: line1\n" in body
|
||||
assert "data: line2\n" in body
|
||||
assert "\r" not in body.split("data: [DONE]")[0]
|
||||
|
||||
def test_streaming_finalize_error_emits_error_frame_no_done(self) -> None:
|
||||
# ``get_final_response()`` is what triggers history-provider
|
||||
# persistence on the agent side; if it fails we must surface that
|
||||
# to the client as ``event: error`` rather than emitting ``[DONE]``
|
||||
# as if the run completed cleanly.
|
||||
class _FailingFinalStream(_FakeStream):
|
||||
async def get_final_response(self) -> _FakeAgentResponse:
|
||||
raise RuntimeError("history backend exploded")
|
||||
|
||||
class _AgentWithFailingFinal(_FakeAgent):
|
||||
def run(self, messages: Any = None, *, stream: bool = False, **kwargs: Any) -> Any:
|
||||
self.calls.append({"messages": messages, "stream": stream, "kwargs": kwargs})
|
||||
if stream:
|
||||
return _FailingFinalStream(["partial"])
|
||||
return super().run(messages, stream=stream, **kwargs)
|
||||
|
||||
agent = _AgentWithFailingFinal()
|
||||
host = AgentFrameworkHost(target=agent, channels=[InvocationsChannel()])
|
||||
with TestClient(host.app) as client:
|
||||
r = client.post("/invocations/invoke", json={"message": "x", "stream": True})
|
||||
assert r.status_code == 200
|
||||
body = r.text
|
||||
assert "data: partial" in body
|
||||
assert "event: error" in body
|
||||
assert "history backend exploded" in body
|
||||
assert "[DONE]" not in body
|
||||
@@ -86,6 +86,7 @@ agent-framework-foundry-local = { workspace = true }
|
||||
agent-framework-gemini = { workspace = true }
|
||||
agent-framework-github-copilot = { workspace = true }
|
||||
agent-framework-hosting = { workspace = true }
|
||||
agent-framework-hosting-invocations = { workspace = true }
|
||||
agent-framework-hyperlight = { workspace = true }
|
||||
agent-framework-lab = { workspace = true }
|
||||
agent-framework-mem0 = { workspace = true }
|
||||
@@ -209,6 +210,7 @@ executionEnvironments = [
|
||||
{ root = "packages/foundry_local/tests", reportPrivateUsage = "none" },
|
||||
{ root = "packages/github_copilot/tests", reportPrivateUsage = "none" },
|
||||
{ root = "packages/hosting/tests", reportPrivateUsage = "none" },
|
||||
{ root = "packages/hosting-invocations/tests", reportPrivateUsage = "none" },
|
||||
{ root = "packages/lab/gaia/tests", reportPrivateUsage = "none" },
|
||||
{ root = "packages/lab/lightning/tests", reportPrivateUsage = "none" },
|
||||
{ root = "packages/lab/tau2/tests", reportPrivateUsage = "none" },
|
||||
|
||||
Generated
+10
-1
@@ -49,6 +49,7 @@ members = [
|
||||
"agent-framework-github-copilot",
|
||||
"agent-framework-hosting",
|
||||
"agent-framework-hosting-responses",
|
||||
"agent-framework-hosting-invocations",
|
||||
"agent-framework-hyperlight",
|
||||
"agent-framework-lab",
|
||||
"agent-framework-mem0",
|
||||
@@ -651,7 +652,6 @@ dependencies = [
|
||||
{ name = "agent-framework-core", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
|
||||
{ name = "agent-framework-hosting", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
|
||||
{ name = "openai", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
|
||||
]
|
||||
|
||||
[package.metadata]
|
||||
requires-dist = [
|
||||
@@ -660,6 +660,15 @@ requires-dist = [
|
||||
{ name = "openai", specifier = ">=1.99.0,<3" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "agent-framework-hosting-invocations"
|
||||
version = "1.0.0a260424"
|
||||
source = { editable = "packages/hosting-invocations" }
|
||||
dependencies = [
|
||||
{ name = "agent-framework-core", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
|
||||
{ name = "agent-framework-hosting", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "agent-framework-hyperlight"
|
||||
version = "1.0.0b260521"
|
||||
|
||||
Reference in New Issue
Block a user