Add agent-langgraph-scratch template (raw FastAPI) - #265
Conversation
A LangGraph agent served from a hand-written FastAPI app (no LongRunningAgentServer / AgentServer). server/app.py is the SDK-agnostic Responses surface (byte-identical to agent-openai-scratch); only the agent/wire handlers differ. In-memory background mode, in-process checkpointer, optional tracing + managed memory store. Co-authored-by: Isaac <no-reply@databricks.com>
Co-authored-by: Isaac <no-reply@databricks.com>
Co-authored-by: Isaac <no-reply@databricks.com>
outbound.py now relays LangChain message dicts + text deltas as-is (no Responses conversion), matching scratch's 'show the SDK's raw shape' intent. Input stays Responses-style. Co-authored-by: Isaac <no-reply@databricks.com>
input list of LangChain message dicts passes straight to the agent; no to_chat_completions_input. scratch now speaks the SDK's native shape on both ends. Co-authored-by: Isaac <no-reply@databricks.com>
Co-authored-by: Isaac <no-reply@databricks.com>
The background-run store is a non-durable stand-in slated for a durable backend, so it belongs with the other mason plumbing. server/app.py imports BackgroundRuns from mason (this diverges its server/app.py from agent-openai-scratch's). Co-authored-by: Isaac <no-reply@databricks.com>
Co-authored-by: Isaac <no-reply@databricks.com>
Co-authored-by: Isaac <no-reply@databricks.com>
Co-authored-by: Isaac <no-reply@databricks.com>
Co-authored-by: Isaac <no-reply@databricks.com>
Co-authored-by: Isaac <no-reply@databricks.com>
Co-authored-by: Isaac <no-reply@databricks.com>
| invoke_handler(request: dict) -> dict | ||
| stream_handler(request: dict) -> AsyncGenerator[dict] | ||
|
|
||
| Nothing here is OpenAI- or LangGraph-specific — the agent SDK lives entirely behind those handlers |
There was a problem hiding this comment.
do we need this section? It reads more like a tutorial than a template
| def __init__(self) -> None: | ||
| self._runs: dict[str, dict[str, Any]] = {} | ||
|
|
||
| def create(self) -> str: |
There was a problem hiding this comment.
nit: unless we plan to separate creation and starting
| def create(self) -> str: | |
| def start(self) -> str: |
| mlflow.update_current_trace(tags={_TRACE_NAME_TAG: "invoke_handler"}) | ||
| span.set_inputs(request) | ||
| result = await invoke_handler(request) | ||
| span.set_attribute(_MESSAGE_FORMAT_ATTR, "openai") |
| outputs = [ | ||
| event["message"] | ||
| async for event in stream_handler(request) | ||
| if event.get("type") == "message" | ||
| ] |
There was a problem hiding this comment.
is delegating invoke to call streaming a standard pattern?
| | Run locally | `uv run start-server` | | ||
| | Test | `uv run pytest` (hermetic; live model test runs only with a profile) | | ||
| | Deploy | `databricks apps deploy agent-langgraph-scratch --source-code-path <path>` | |
There was a problem hiding this comment.
will these be updated with the mason CLI once it's ready?
| app.add_api_route("/invocations", _handle, methods=["POST"]) | ||
| app.add_api_route("/responses", _handle, methods=["POST"]) |
There was a problem hiding this comment.
do we need both routes? what if we only keep invocations?
| def complete(self, response_id: str, output: dict) -> None: | ||
| self._runs[response_id] = {"status": "completed", "output": output, "error": None} | ||
|
|
||
| def fail(self, response_id: str, error: str) -> None: | ||
| self._runs[response_id] = {"status": "failed", "output": None, "error": error} | ||
|
|
||
| def get(self, response_id: str) -> dict | None: |
There was a problem hiding this comment.
will using response_id be confusing if the runtime service API may not be Responses API compatible?
| app.add_api_route("/invocations", _handle, methods=["POST"]) | ||
| app.add_api_route("/responses", _handle, methods=["POST"]) | ||
|
|
||
| @app.get("/responses/{response_id}") |
There was a problem hiding this comment.
should this be changed to /invocations/{invocation_id} to better decouple from Responses?
There was a problem hiding this comment.
this wire module is a misnomer since it's less about request/response transformation which is what's here vs raw encoding. These are pretty thin right now, can we move them into app.py?
| def get_session_id(request: dict) -> str: | ||
| """Return the request's ``session_id`` (for multi-turn), or a fresh UUID for a new conversation.""" | ||
| return str(request.get("session_id") or uuid7()) |
There was a problem hiding this comment.
let's update this to pull the session_id from the header as our recent discussion: https://databricks.slack.com/archives/C0BL69HA4DQ/p1787345723987959
| # Leave UNSET to skip tracing (local dev). To enable, set a destination AND an experiment (either | ||
| # form of each works): destination = MLFLOW_TRACKING_URI or MLFLOW_TRACING_DESTINATION; | ||
| # experiment = MLFLOW_EXPERIMENT_ID or MLFLOW_EXPERIMENT_NAME. | ||
| # MLFLOW_TRACKING_URI="databricks" | ||
| # MLFLOW_EXPERIMENT_ID= | ||
| # MLFLOW_EXPERIMENT_NAME= |
There was a problem hiding this comment.
it should be possible to create a local mlflow tracking server from the OSS code if we want this to work E2E in dev mode, let's investigate that. (Not a blocker for this PR)
jamesbxwu
left a comment
jamesbxwu
left a comment
There was a problem hiding this comment.
discussed offline, can we move these templates for now into databricks-ai-bridge/mason/templates until we are ready to publish them?
jamesbxwu
left a comment
jamesbxwu
left a comment
There was a problem hiding this comment.
suggestion on folder structure: can we rename the top level server folder to runtime and potentially the following files
- start_server.py -> main.py
- app.py -> runtime.py
2 participants
A LangGraph agent served from a hand-written FastAPI app (no LongRunningAgentServer / AgentServer). server/app.py is the SDK-agnostic Responses surface (byte-identical to agent-openai-scratch); only the agent/wire handlers differ. In-memory background mode, in-process checkpointer, optional tracing + managed memory store.