Build a Foreground App

Expose MCP tools the Truffle agent can call on demand

Foreground apps are MCP servers that give the Truffle agent callable tools. You subclass ForegroundApp, register tools with a decorator, and call app.run().

Foreground apps serve MCP over streamable HTTP. stdio transport is not supported for deployed apps.

If you’re new to MCP, start with the protocol docs.

---

Foreground patterns in the examples

There are two common ways to build a foreground app:

Pattern	Use it when	Examples
Native ForegroundApp	You own the tool functions and can shape inputs/outputs directly.	arxiv, exa, home-assistant, obsidian, whoop
Remote MCP wrapper/proxy	An upstream MCP server already exists, but you need auth, metadata cleanup, argument normalization, safety checks, or compact result formatting.	notion, viator

Prefer the native ForegroundApp pattern when you are writing the integration yourself. Use the remote MCP wrapper pattern when porting an existing MCP server or when the provider hosts an MCP endpoint.

---

Minimal example

A complete, working foreground app that exposes one tool:

# my_app_foreground.py
from __future__ import annotations

import atexit
import httpx
from mcp.types import CallToolResult, TextContent

from truffile.app_runtime import ForegroundApp, ToolSpec, err


class MyApp(ForegroundApp):
    def __init__(self) -> None:
        super().__init__("myapp", logger_name="myapp.foreground")
        self._http = httpx.AsyncClient(timeout=30.0)
        self._register_tools()

    def _register_tools(self) -> None:
        @self.tool(
            ToolSpec(
                name="search_web",
                description=(
                    "Search the web and return the top results. "
                    "Use this when the user asks about current events or recent information."
                ),
                icon="magnifying-glass",
                annotations={"readOnlyHint": True, "destructiveHint": False},
            )
        )
        async def search_web(query: str, limit: int = 5) -> CallToolResult | dict:
            """Search the web for `query` and return up to `limit` results."""
            if not query.strip():
                return err("query is required")
            limit = max(1, min(limit, 20))
            try:
                response = await self._http.get(
                    "https://api.example.com/search",
                    params={"q": query, "n": limit},
                )
                response.raise_for_status()
                results = response.json().get("results", [])[:limit]
                lines = [f"### Search results for `{query}`"]
                for i, item in enumerate(results, 1):
                    title = item.get("title", "Untitled")
                    url = item.get("url", "")
                    snippet = item.get("snippet", "")
                    lines.append(f"- [{i}] {title} — {url}")
                    if snippet:
                        lines.append(f"  {snippet[:240]}")
                return CallToolResult(
                    content=[TextContent(type="text", text="\n".join(lines))],
                    structuredContent={
                        "query": query,
                        "results": [
                            {
                                "title": item.get("title", ""),
                                "url": item.get("url", ""),
                            }
                            for item in results
                        ],
                    },
                )
            except httpx.HTTPStatusError as exc:
                return err(f"HTTP {exc.response.status_code}", status_code=exc.response.status_code)

    async def aclose(self) -> None:
        await self._http.aclose()


app = MyApp()


def _cleanup() -> None:
    try:
        import asyncio
        asyncio.run(app.aclose())
    except Exception:
        pass


atexit.register(_cleanup)


if __name__ == "__main__":
    app.run()

That’s the whole pattern. The rest of this page explains each piece.

---

The ForegroundApp class

Subclass ForegroundApp and call super().__init__(name, logger_name=...). The constructor sets up the MCP server machinery, your logger (self.logger), and the internal tool registry.

class MyApp(ForegroundApp):
    def __init__(self) -> None:
        super().__init__("myapp", logger_name="myapp.foreground")

• name — a short identifier for your app, used in MCP server metadata.
• logger_name — optional; defaults to the app name. Use a namespaced logger so your logs are distinguishable in the runtime.

---

Registering tools

Tools are async functions registered via the @self.tool(...) decorator. The decorator takes a ToolSpec with the tool’s name, description, and icon:

from truffile.app_runtime import ToolSpec

@self.tool(
    ToolSpec(
        name="get_repo",
        description=(
            "Fetch a GitHub repository's metadata by owner and repo name. "
            "Returns stars, default branch, and description."
        ),
        icon="github-logo",
        annotations={"readOnlyHint": True, "destructiveHint": False},
    )
)
async def get_repo(owner: str, repo: str) -> dict:
    ...

Fields on ToolSpec:

Field	Meaning
name	The tool name the model sees. Use snake_case.
description	Plain English. The model uses this to decide when to call the tool — make it explicit.
icon	Either a full URL or a Phosphor icon name (e.g. "search", "github-logo").
title	Optional human-readable label. Defaults to name.
annotations	Optional MCP annotations. Use {"readOnlyHint": True} for read-only tools and {"destructiveHint": True} for tools that delete, send, purchase, publish, or otherwise mutate important state.
structured_output	Optional FastMCP structured-output flag. Most apps can leave this unset.

Icons

Passing a name like "search" is equivalent to calling phosphor_icon_url("search") — the SDK turns it into a Phosphor SVG URL. If you want a custom icon, pass the full URL instead:

ToolSpec(name="...", description="...", icon="https://example.com/icons/myicon.svg")

Type hints are your tool schema

The tool function’s type hints become the MCP tool’s input schema. Keep them concrete:

async def search(query: str, limit: int = 5, include_archived: bool = False) -> dict:
    ...

Supported types: str, int, float, bool, list[...], dict[...], Optional[...]. Give every parameter a sensible default where possible — the model can call the tool with fewer arguments if you do.

---

Wrapping a remote MCP server

Some apps are mostly adapters around another MCP server. The wrapper still matters because it is where you make the provider useful and safe for the Truffle agent:

• add or override tool titles, descriptions, icons, and annotations
• normalize provider-specific argument names into model-friendly arguments
• strip debug-only fields from tool results
• convert verbose provider JSON into compact markdown plus small structuredContent
• report auth/runtime errors through Truffle’s app error path

The notion example is a remote MCP proxy with OAuth:

class NotionForegroundApp:
    def list_tools(self) -> list[dict[str, Any]]:
        client = self.get_client()
        tools = client.list_mcp_tools()
        return [_enrich_tool_metadata(tool) for tool in tools]

    async def invoke_tool(self, name: str, **arguments: Any) -> Any:
        try:
            return self.get_client().call_tool(name, arguments)
        except Exception as exc:
            await self._error_reporter.report_foreground_exception(exc, tool_name=name)
            raise

The viator example goes further: it wraps a remote MCP server, overrides the two public tool schemas, adds local arguments like include_raw, and formats compact travel-search results before returning them.

Use this pattern when an upstream MCP is already good enough to call, but not good enough to expose raw.

---

Returning results

For simple action tools, returning a small dict with ok() / err() is fine:

from truffile.app_runtime import ok, err

return ok("Created draft", draft_id=draft_id)

return err("Rate limit exceeded", retry_after=60)

For read tools that can return many rows, prefer an MCP CallToolResult with:

• concise markdown in content, optimized for the model to read
• small structuredContent, only for IDs or fields the model may need for follow-up calls
• no duplicated status/tool-name fields
• no raw provider payloads unless the user explicitly asks for a detailed/debug view

from mcp.types import CallToolResult, TextContent

return CallToolResult(
    content=[TextContent(type="text", text="### Issues\n- [1] #42 Fix deploy cleanup")],
    structuredContent={
        "issues": [{"ref": 1, "number": 42, "id": "I_kw..."}],
    },
)

Plain dict returns are serialized as JSON for the model. That is convenient, but it gets token-expensive when the dict is large. If a tool returns lists, messages, search results, documents, or other verbose records, write a compact text summary yourself.

For remote MCP wrappers, keep the same rule: forward the provider result only when it is already compact. Otherwise, build the model-facing text yourself and keep structuredContent to stable IDs, refs, pagination, or fields required for follow-up calls.

---

Error handling

For errors you expect (bad inputs, HTTP failures, rate limits), return err(...). For authentication failures — where the user needs to reinstall or refresh credentials — raise AppAuthError:

from truffile.app_runtime import AppAuthError, AppRuntimeFailure, err

async def list_issues(owner: str, repo: str) -> dict:
    try:
        response = await self._http.get(f"/repos/{owner}/{repo}/issues")
        response.raise_for_status()
        return ok("Listed issues", issues=response.json())
    except httpx.HTTPStatusError as exc:
        if exc.response.status_code in (401, 403):
            raise AppAuthError("GitHub token is invalid or lacks permissions") from exc
        return err(f"GitHub returned HTTP {exc.response.status_code}")
    except httpx.HTTPError as exc:
        raise AppRuntimeFailure(f"GitHub request failed: {exc}") from exc

• AppAuthError — the runtime classifies this as an auth problem and flags the app as needing user intervention. Use it for 401/403 responses, expired tokens, or missing credentials.
• AppRuntimeFailure — for generic runtime issues you want to propagate instead of returning as a soft error.
• Returning err(...) — for expected errors the agent should see and potentially recover from (bad user input, rate limits, upstream 5xx, etc.).

---

Tool design guidance

A few things we’ve learned from building the reference apps:

1. Validate inputs early. If the caller passes limit=99999, clamp it. If an enum arg is invalid, return err(...) with the valid values.
2. Optimize result text for the model. For read tools, concise markdown beats raw provider JSON. Keep structured payloads small and action-oriented.
3. Write descriptions the model can route on. “Search X for Y. Use this when Z.” Describe when to use the tool, not just what it does.
4. Keep one concept per tool. get_market and create_order are better than a single market_op with a mode arg.
5. Name parameters clearly. ticker not t, include_text not text.

---

Testing with truffile infer

You can point the inference REPL at your local MCP server and exercise it end-to-end before deploying.

1. Start your app locally:

   python my_app_foreground.py
   

   It’ll listen on http://127.0.0.1:8000/mcp by default.

2. In another terminal, open the infer REPL:

   truffile infer
   

3. Connect to your MCP server and list the tools:

   > /mcp connect http://127.0.0.1:8000/mcp
   /mcp connect ✓ http://127.0.0.1:8000/mcp (2 tools)
   
   > /mcp tools
   /mcp tools search_web, get_repo
   

4. Run a real prompt that should trigger one of your tools:

   > use the search_web tool to find recent news about SpaceX
   

This is the fastest way to check that your tool descriptions actually route correctly before you deploy.

Never commit real API keys in docs, screenshots, logs, or recordings.

---

Cleanup on shutdown

Close any HTTP clients, database connections, or subprocesses on exit:

import atexit

def _cleanup() -> None:
    try:
        # close sync clients directly, async clients via asyncio.run
        client.close()
    except Exception:
        pass

atexit.register(_cleanup)

Skipping this is the #1 cause of flaky redeploys and container runtime crashes.

---

Deploy

truffile validate ./my-app
truffile deploy --dry-run ./my-app
truffile deploy ./my-app

After deployment, send a one-shot Convo smoke request:

truffile convo --new --json --quiet --timeout 120 \
  "Use <tool behavior> and summarize the result"

Inspect tool_calls in the JSON result. The model may discover the installed tool through normal routing, but Convo cannot attach or restrict an app per chat, so this is not deterministic enforcement. Before deployment, use truffile infer --mcp <url> when you need to select and test the local MCP server directly.

---

When to build a foreground app

Build a foreground app when you need:

• Request/response tools the agent calls during a Convo request (search, fetch, lookup, CRUD operations)
• Deterministic API integrations the model should reach for in specific situations
• Action-style behaviors (post, create, cancel, trigger)

If instead you want something that runs on a schedule and feeds context to the agent without the user asking, you want a background app — or a hybrid app that has both.