feat(server): FastAPI SSE endpoint + Gemini LLM support

Author: HalukProductions

mini_agent_harness/cli.py

@@ -31,9 +31,19 @@ def init(agent_name: Optional[str] = typer.Argument("quickstart", help="Name of
 @app.command()
 def serve(model: str = typer.Option("gpt-3.5-turbo", "--model", help="Model name or path to serve with")):
     """Serve the agent via FastAPI (placeholder implementation)."""
-    typer.echo(
-        "🚧 The serve command is not implemented yet. Run `mini-agent init` and explore tests in the meantime."
-    )
+    try:
+        import uvicorn  # type: ignore
+    except ModuleNotFoundError as exc:  # pragma: no cover
+        typer.echo("Error: 'uvicorn' not installed. Run `poetry add uvicorn fastapi --group main`.")
+        raise typer.Exit(1) from exc
+
+    from importlib import import_module
+
+    # Import here to avoid FastAPI requirement unless serve is run
+    server_mod = import_module("mini_agent_harness.server")
+
+    typer.echo(f"🚀 Serving on http://127.0.0.1:8000  (model={model})")
+    uvicorn.run(server_mod.app, host="0.0.0.0", port=8000, reload=False)
 
 
 @app.command()

mini_agent_harness/core/llm.py

@@ -49,24 +49,61 @@ def generate(self, prompt: str) -> str:  # pragma: no cover
         return completion.choices[0].message.content
 
 
+# ---------------- Google Gemini -----------------
+
+
+class GeminiLLM:  # pragma: no cover
+    """Wrapper around Google Generative AI Gemini models."""
+
+    def __init__(self, model: str = "gemini-pro") -> None:
+        try:
+            import google.generativeai as genai  # type: ignore
+        except ModuleNotFoundError:  # pragma: no cover
+            raise ImportError(
+                "google-generativeai not installed. Run `poetry add google-generativeai`."
+            ) from None
+
+        api_key = os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY")
+        if not api_key:
+            raise RuntimeError("GEMINI_API_KEY environment variable not set.")
+
+        genai.configure(api_key=api_key)
+        self._model = genai.GenerativeModel(model)
+
+    def generate(self, prompt: str) -> str:  # noqa: D401
+        response = self._model.generate_content(prompt)
+        # Newer client returns text in .text
+        return response.text  # type: ignore[attr-defined]
+
+
 _DEF_PROVIDER = "echo"  # default provider if nothing specified
 
+_PROVIDERS: dict[str, type[LLM]] = {
+    "openai": OpenAILLM,
+    "gemini": GeminiLLM,
+    "echo": EchoLLM,
+}
+
 
 def get_default_llm() -> LLM:
     """Return an LLM instance based on env vars.
 
     Priority order:
     1. `MINI_AGENT_LLM` env var ("openai" or "echo").
     2. If `OPENAI_API_KEY` is set → "openai".
-    3. Fallback to "echo".
+    3. If `GEMINI_API_KEY` or `GOOGLE_API_KEY` is set → "gemini".
+    4. Fallback to "echo".
     """
 
-    provider = os.getenv("MINI_AGENT_LLM") or (
-        "openai" if os.getenv("OPENAI_API_KEY") else _DEF_PROVIDER
-    )
+    provider = os.getenv("MINI_AGENT_LLM")
 
-    if provider == "openai":
-        return OpenAILLM()
+    if not provider:
+        if os.getenv("OPENAI_API_KEY"):
+            provider = "openai"
+        elif os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY"):
+            provider = "gemini"
+        else:
+            provider = _DEF_PROVIDER
 
-    # Default to echo
-    return EchoLLM() 
+    cls = _PROVIDERS.get(provider, EchoLLM)
+    return cls() 

mini_agent_harness/server.py

@@ -0,0 +1,46 @@
+"""FastAPI server exposing chat endpoint with simple streaming.
+
+Run via `mini-agent serve` (see CLI). This is an early skeleton: it
+instantiates an Agent for each request using the quickstart manifest and
+streams back text tokens separated by spaces.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import AsyncGenerator, Iterator
+
+import yaml # type: ignore
+from fastapi import FastAPI, Form # type: ignore
+from fastapi.responses import StreamingResponse # type: ignore
+
+from .core import Agent
+
+app = FastAPI(title="MiniAgentHarness")
+
+_MANIFEST_PATH = Path("agents/quickstart.yaml")
+
+
+def _get_agent() -> Agent:
+    manifest = yaml.safe_load(_MANIFEST_PATH.read_text())
+    return Agent(manifest)
+
+
+def _stream_text(text: str) -> Iterator[bytes]:
+    for token in text.split():
+        yield f"data: {token}\n\n".encode()
+
+
+@app.get("/")
+async def root() -> dict[str, str]:
+    return {"status": "ok"}
+
+
+@app.post("/chat")
+async def chat(message: str = Form(...)):  # noqa: D401
+    agent = _get_agent()
+    result = agent.run(message)
+    return StreamingResponse(
+        _stream_text(result.response_text),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache"},
+    ) 

mini_agent_harness/tools/__init__.py

@@ -19,7 +19,7 @@
 from pathlib import Path
 from typing import Callable, Dict, Iterable, List
 
-import yaml 
+import yaml  # type: ignore
 
 
 @dataclass