Add LangChain vm_run wrapper example
This commit is contained in:
parent
f7c8a4366b
commit
a5cde71b37
4 changed files with 153 additions and 0 deletions
84
examples/langchain_vm_run.py
Normal file
84
examples/langchain_vm_run.py
Normal file
|
|
@ -0,0 +1,84 @@
|
|||
"""Thin LangChain wrapper over the public vm_run contract.
|
||||
|
||||
Requirements:
|
||||
- `pip install langchain-core` or `uv add langchain-core`
|
||||
|
||||
This example keeps the framework integration intentionally narrow: a single tool
|
||||
that delegates straight to `Pyro.run_in_vm(...)`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Any, Callable, TypeVar, cast
|
||||
|
||||
from pyro_mcp import Pyro
|
||||
|
||||
F = TypeVar("F", bound=Callable[..., Any])
|
||||
|
||||
|
||||
def run_vm_run_tool(
|
||||
*,
|
||||
profile: str,
|
||||
command: str,
|
||||
vcpu_count: int,
|
||||
mem_mib: int,
|
||||
timeout_seconds: int = 30,
|
||||
ttl_seconds: int = 600,
|
||||
network: bool = False,
|
||||
) -> str:
|
||||
pyro = Pyro()
|
||||
result = pyro.run_in_vm(
|
||||
profile=profile,
|
||||
command=command,
|
||||
vcpu_count=vcpu_count,
|
||||
mem_mib=mem_mib,
|
||||
timeout_seconds=timeout_seconds,
|
||||
ttl_seconds=ttl_seconds,
|
||||
network=network,
|
||||
)
|
||||
return json.dumps(result, sort_keys=True)
|
||||
|
||||
|
||||
def build_langchain_vm_run_tool() -> Any:
|
||||
try:
|
||||
from langchain_core.tools import tool # type: ignore[import-not-found]
|
||||
except ImportError as exc: # pragma: no cover - dependency guard
|
||||
raise RuntimeError(
|
||||
"langchain-core is required for this example. Install it with "
|
||||
"`uv add langchain-core` or `pip install langchain-core`."
|
||||
) from exc
|
||||
|
||||
decorator = cast(Callable[[F], F], tool("vm_run"))
|
||||
|
||||
@decorator
|
||||
def vm_run(
|
||||
profile: str,
|
||||
command: str,
|
||||
vcpu_count: int,
|
||||
mem_mib: int,
|
||||
timeout_seconds: int = 30,
|
||||
ttl_seconds: int = 600,
|
||||
network: bool = False,
|
||||
) -> str:
|
||||
"""Run one command in an ephemeral Firecracker VM and clean it up."""
|
||||
return run_vm_run_tool(
|
||||
profile=profile,
|
||||
command=command,
|
||||
vcpu_count=vcpu_count,
|
||||
mem_mib=mem_mib,
|
||||
timeout_seconds=timeout_seconds,
|
||||
ttl_seconds=ttl_seconds,
|
||||
network=network,
|
||||
)
|
||||
|
||||
return vm_run
|
||||
|
||||
|
||||
def main() -> None:
|
||||
tool = build_langchain_vm_run_tool()
|
||||
print(tool)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue