Building a Terminal Agent with Venice AI and Apify

A model on its own cannot tell you what is on the front page of Hacker News right now. To do that, it needs tools, and someone has to build and maintain those tools. Apify already did: it hosts thousands of Actors that scrape sites, crawl documentation, and pull structured data, and it exposes them over the Model Context Protocol. That combination is a good fit for Venice. Venice supplies OpenAI-compatible function calling with no data retention, Apify supplies the tools, and MCP is the wire format between them. You do not write a scraper per site — you connect once and let the model pick the Actor. In this tutorial, we’ll build a terminal agent in Python that does exactly that. By the end, you’ll have a CLI that discovers a Venice function-calling model at runtime, loads the Apify tool catalog over MCP, streams answers into your terminal, and asks before it spends money on an Actor run. Interested in the full code implementation? Check out the GitHub repo. Before we continue, you’ll need a Venice API key:

export VENICE_API_KEY=<my-key>

What We’re Building

The reference implementation is a small Python package with one job per module:

Module	What it does
`config.py`	Loads settings from the environment or `.env`, and builds the Apify MCP URL
`venice.py`	Resolves a model from `GET /models/traits` and streams chat completions
`apify_mcp.py`	Owns the MCP transport and calls Apify tools
`tools.py`	Converts MCP tool schemas into Venice function tools and formats results
`agent.py`	Runs the tool-calling loop and gates paid tools
`cli.py`	Typer entry point, REPL, slash commands, and approval prompts
`render.py`	Rich output for the banner, streamed tokens, and tool calls

A single question flows through it like this:

Ask Venice for the current function-calling model, unless you pinned one.
Connect to the Apify MCP server and list its tools.
Rewrite those MCP tools as OpenAI-compatible function definitions.
Send the question with the tool list attached.
If the model returns tool_calls, run them against Apify and append the results as tool messages.
Repeat until the model answers with text instead of a tool call.

Steps 4 through 6 are the whole agent. Everything else exists to make those three steps safe and pleasant to use.

This agent can spend Apify compute on your account. Start without APIFY_TOKEN if you only want search and documentation tools, and leave --yes off until you actually intend to run Actors.

Setting Up the Project

The reference project uses Python 3.12+ and uv. Create a new project:

mkdir venice-terminal-agent
cd venice-terminal-agent
uv init --package

Install the dependencies:

uv add httpx2 mcp openai prompt-toolkit pydantic-settings python-dotenv rich typer
uv add --dev pytest pytest-asyncio

That is httpx2, the 2.x line of httpx, which both openai and mcp already depend on. Installing it directly avoids ending up with two HTTP clients in the same environment. Then create a .env file:

VENICE_API_KEY=your_venice_api_key_here
APIFY_TOKEN=your_apify_token_here

VENICE_API_KEY comes from Venice API settings. APIFY_TOKEN comes from Apify Console and is optional — we’ll cover what you get without it in a moment.

Loading Configuration

Settings come first because every other module takes them as an argument. We’ll use pydantic-settings so environment variables, .env, and CLI flags all land in one validated object. In src/venice_terminal_agent/config.py, a Settings(BaseSettings) class carries the fields that matter:

class Settings(BaseSettings):
    venice_api_key: str = Field(min_length=1)
    venice_model: str | None = None

    apify_token: str | None = None
    apify_mcp_transport: Literal["http", "stdio"] = "http"
    apify_mcp_tools: str | None = None

    max_rounds: int = Field(default=12, ge=1, le=40)
    max_tool_result_chars: int = Field(default=20_000, ge=1_000)
    # base URLs, temperature, auto_approve_tools, and save_history omitted

Two of these carry decisions rather than defaults. venice_model is None rather than a model ID, which we’ll come back to in the next section. And max_rounds with max_tool_result_chars are the limits that stop an agent running away: the first caps how many tool rounds one question can take, the second caps how much of a scraped page gets fed back into context. The interesting function in this module is the URL builder:

ANONYMOUS_APIFY_TOOLS = (
    "search-actors,fetch-actor-details,search-apify-docs,fetch-apify-docs"
)


def apify_http_url(settings: Settings) -> str:
    base = settings.apify_mcp_url.rstrip("/")
    tools = settings.apify_mcp_tools
    if tools is None and not settings.apify_token:
        tools = ANONYMOUS_APIFY_TOOLS
    if tools:
        separator = "&" if "?" in base else "?"
        return f"{base}{separator}tools={tools}"
    return base

The hosted Apify MCP server takes a tools query parameter that decides which tools it advertises. Without an APIFY_TOKEN we ask for the four anonymous tools that work unauthenticated — Actor search, Actor details, documentation search, and documentation fetch. That means someone can clone the project, add only a Venice key, and still get a working agent that can research Apify Actors. They just cannot run one.

Talking to Venice

Venice is OpenAI-compatible, so we can use the OpenAI SDK for chat completions and plain httpx for the model-discovery call. Create src/venice_terminal_agent/venice.py:

from __future__ import annotations

from collections.abc import Callable
from typing import Any

import httpx2
from openai import APIStatusError, AsyncOpenAI

from venice_terminal_agent.config import Settings

CHAT_TIMEOUT_SECONDS = 180.0


class VeniceClient:
    """Thin wrapper around Venice's OpenAI-compatible chat API."""

    def __init__(self, settings: Settings) -> None:
        self._settings = settings
        self._http = httpx2.AsyncClient(
            base_url=settings.venice_base_url.rstrip("/"),
            headers={"Authorization": f"Bearer {settings.venice_api_key}"},
            timeout=30.0,
            follow_redirects=True,
        )
        self._chat = AsyncOpenAI(
            api_key=settings.venice_api_key,
            base_url=settings.venice_base_url.rstrip("/"),
            timeout=CHAT_TIMEOUT_SECONDS,
        )

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

Two clients for one API looks redundant, but they do different jobs. AsyncOpenAI gives us the streaming helper and typed tool_calls for free. The raw httpx client is there for Venice endpoints the OpenAI SDK does not know about, which in this project means /models/traits. The chat timeout is much longer than the discovery timeout on purpose. A question that triggers a web crawl can legitimately take a couple of minutes.

Discovering a Model at Runtime

Venice model IDs rotate, and hardcoding one is the fastest way to ship an agent that breaks in a month. GET /models/traits maps stable trait names onto whatever model currently fills that role, so we ask for function_calling_default instead of naming a model:

    async def resolve_model(self, override: str | None = None) -> str:
        if override:
            return override
        if self._settings.venice_model:
            return self._settings.venice_model
        response = await self._http.get("/models/traits", params={"type": "text"})
        try:
            response.raise_for_status()
        except httpx2.HTTPStatusError as error:
            raise RuntimeError(format_http_error(error)) from error
        data = response.json().get("data") or {}
        model = data.get("function_calling_default") or data.get("default")
        if not model:
            raise RuntimeError("Venice /models/traits did not return a function-calling model")
        return str(model)

The precedence here matters: an explicit --model flag wins, then VENICE_MODEL from the environment, then the trait lookup. So the default path needs no configuration at all, but you can still pin a model when you are comparing behaviour between two of them.

Not every text model supports function calling. Asking for the function_calling_default trait means you get one that does, without maintaining a list yourself. See Deprecations for how often the underlying IDs change.

Streaming Completions

Now add the completion call:

    async def complete(
        self,
        *,
        model: str,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]] | None,
        on_text: Callable[[str], None] | None = None,
    ) -> Any:
        kwargs = chat_request_kwargs(
            model=model,
            messages=messages,
            tools=tools,
            temperature=self._settings.venice_temperature,
        )
        async with self._chat.chat.completions.stream(**kwargs) as stream:
            async for event in stream:
                if on_text is not None and event.type == "content.delta" and event.delta:
                    on_text(event.delta)
            completion = await stream.get_final_completion()
        return completion.choices[0].message

We stream so the user sees text appear as it is generated, but we still want the assembled message afterwards — tool calls arrive in fragments across many chunks, and reassembling them by hand is tedious. The SDK’s stream() context manager handles both: content.delta events drive the terminal output, and get_final_completion() hands back a complete message with tool_calls already stitched together. The request itself is built by a separate function so it stays easy to test:

def chat_request_kwargs(
    *,
    model: str,
    messages: list[dict[str, Any]],
    tools: list[dict[str, Any]] | None,
    temperature: float,
) -> dict[str, Any]:
    kwargs: dict[str, Any] = {
        "model": model,
        "messages": messages,
        "temperature": temperature,
        "extra_body": {
            "venice_parameters": {
                "include_venice_system_prompt": False,
            }
        },
    }
    if tools:
        kwargs["tools"] = tools
        kwargs["tool_choice"] = "auto"
    return kwargs

extra_body is how the OpenAI SDK passes through fields it does not model, which is where venice_parameters goes. Setting include_venice_system_prompt to false keeps Venice’s default assistant prompt out of the conversation, so our own system prompt is the only instruction the model gets. For an agent with strict tool rules, that is what you want. Only attach tools and tool_choice when there is at least one tool. Sending an empty tools array is a needless way to confuse a model. The module also has a format_http_error() helper that turns an APIStatusError or an httpx2.HTTPStatusError into a one-line string with the status code and response body. Agents fail at the API boundary more often than anywhere else, and a readable message there saves a lot of guessing.

Converting MCP Tools into Venice Tools

MCP tools and OpenAI-style function tools describe the same thing in different shapes. Both have a name, a description, and a JSON Schema for arguments. The translation is mostly mechanical, with one catch: Apify tool names include characters that function names do not allow. An Actor tool might be called apify/rag-web-browser, and that slash is not valid. So we sanitize the names on the way out and keep a map so we can restore them on the way back in. In src/venice_terminal_agent/tools.py, a ToolCatalog does the translation and holds the map:

class ToolCatalog:
    """Maps Venice-safe function names back to MCP tool names."""

    def __init__(self, tools: Iterable[Any]) -> None:
        self.openai_tools: list[dict[str, Any]] = []
        self._mcp_names: dict[str, str] = {}
        used: set[str] = set()
        for tool in tools:
            raw_name = str(getattr(tool, "name", "") or "tool")
            safe_name = unique_name(sanitize_tool_name(raw_name), used)
            used.add(safe_name)
            self._mcp_names[safe_name] = raw_name
            self.openai_tools.append(
                {
                    "type": "function",
                    "function": {
                        "name": safe_name,
                        "description": str(getattr(tool, "description", "") or raw_name),
                        "parameters": tool_input_schema(tool),
                    },
                }
            )

    def mcp_name(self, venice_name: str) -> str:
        return self._mcp_names.get(venice_name, venice_name)

    def names(self) -> list[str]:
        return [item["function"]["name"] for item in self.openai_tools]

Three small helpers do the unglamorous work. sanitize_tool_name() replaces illegal characters with hyphens, prefixes names that start with a digit, and truncates to 64 characters. unique_name() then appends a numeric suffix if that truncation made two Actors collide — which saves you a genuinely confusing bug where the model calls one Actor and a different one runs. tool_input_schema() copes with MCP servers handing back a dict, a Pydantic model, or nothing at all.

Formatting Results Back into Context

Tool results go straight into the conversation, so they need to be a string, and they need a size limit. Scraping a documentation site can easily return more text than the context window holds. format_tool_result() prefers structured_content when the server provides it, and otherwise flattens the content blocks into text, coping with blocks that are not TextContent. It ends with the two lines that matter:

    if getattr(result, "is_error", False):
        text = json.dumps({"error": text}, ensure_ascii=False)
    return truncate_text(text, max_chars)


def truncate_text(text: str, max_chars: int) -> str:
    if len(text) <= max_chars:
        return text
    omitted = len(text) - max_chars
    return (
        f"{text[:max_chars]}\n\n[truncated {omitted} characters; "
        "use a follow-up tool call with filters, limit, or offset]"
    )

The truncation notice is written for the model, not for you. Telling it that content was cut and suggesting filters, limits, or offsets is usually enough for it to make a narrower second call instead of assuming it saw everything. Errors get wrapped as {"error": "..."} rather than raised. A failed tool call is information the model can act on — it can pick a different Actor or fix its arguments — and it can only do that if the failure reaches it as a normal tool result.

Marking the Tools That Cost Money

Apify tools split cleanly into two groups: ones that read metadata and documentation, and ones that start compute. We want confirmation for the second group, so we allowlist the first:

READ_ONLY_APIFY_TOOLS = frozenset(
    {
        "search-actors",
        "fetch-actor-details",
        "search-apify-docs",
        "fetch-apify-docs",
        "get-actor-output",
        # plus the other get-actor-*, get-dataset-*, and get-key-value-store-* tools
    }
)


def requires_confirmation(name: str) -> bool:
    """True for tools that can spend Apify compute, such as Actor runs."""

    normalized = name.strip().lower().replace("_", "-")
    return normalized not in READ_ONLY_APIFY_TOOLS

An allowlist rather than a blocklist is the important choice. Apify keeps adding tools and Actors, and anything the agent has not seen before defaults to asking first. Get this backwards and every new Actor is auto-approved.

Connecting to Apify over MCP

Apify offers two ways in. The hosted server at https://mcp.apify.com speaks Streamable HTTP, and @apify/actors-mcp-server runs locally over stdio via npx. We’ll support both, since they suit different situations: hosted needs no Node.js, and stdio keeps the connection on your own machine. In src/venice_terminal_agent/apify_mcp.py, an ApifyMcp class wraps the connected session. Its call_tool() is where the sanitized name gets translated back — Venice sends apify-rag-web-browser, Apify receives apify/rag-web-browser:

    async def call_tool(self, venice_name: str, arguments: dict[str, Any]) -> str:
        mcp_name = self.catalog.mcp_name(venice_name)
        result = await self._client.call_tool(mcp_name, arguments)
        return format_tool_result(result, max_chars=self._max_result_chars)

Building the catalog needs a cursor loop over client.list_tools(), since a token with access to many Actors produces a paginated list.

Owning the Transport

An MCP connection is a long-lived async resource, and so is the HTTP client underneath it. An ApifyMcpSession async context manager holds both in an AsyncExitStack, picks a transport based on settings, and loads the catalog. The detail worth copying is the cleanup:

    async def __aenter__(self) -> ApifyMcp:
        try:
            # connect with _connect_stdio or _connect_http, then load_catalog(client)
            return self.session
        except BaseException:
            await self._stack.aclose()
            raise

That except BaseException matters more than it looks. If listing tools fails after the transport is up, without it you leak a subprocess or an open socket every time the agent fails to start. Here are the two transports:

    async def _connect_http(self, settings: Settings) -> Client:
        headers: dict[str, str] = {}
        if settings.apify_token:
            headers["Authorization"] = f"Bearer {settings.apify_token}"
        http = await self._stack.enter_async_context(
            httpx2.AsyncClient(
                headers=headers,
                timeout=httpx2.Timeout(30.0, read=300.0),
                follow_redirects=True,
            )
        )
        transport = streamable_http_client(apify_http_url(settings), http_client=http)
        return await self._stack.enter_async_context(Client(transport))

    async def _connect_stdio(self, settings: Settings) -> Client:
        if not settings.apify_token:
            raise RuntimeError("APIFY_TOKEN is required for the local stdio Apify MCP server")
        args = ["-y", "@apify/actors-mcp-server"]
        if settings.apify_mcp_tools:
            args.extend(["--tools", settings.apify_mcp_tools])
        params = StdioServerParameters(
            command="npx",
            args=args,
            env={"APIFY_TOKEN": settings.apify_token},
        )
        return await self._stack.enter_async_context(Client(stdio_client(params)))

Note the read timeout of 300 seconds on the HTTP transport. Actor runs are slow, and the default 30-second timeout will cut off perfectly healthy crawls. Note also that the stdio subprocess gets only APIFY_TOKEN in its environment, not your whole shell environment — including your Venice key.

Executing a Tool Call

The last piece of this module, execute_venice_tool_call(), turns a Venice tool call into a string result. It wraps both classes of failure — unparseable arguments and a failed Apify call — as {"error": "..."} rather than raising:

    try:
        arguments = parse_tool_arguments(arguments_json)
    except (ValueError, TypeError, json.JSONDecodeError) as error:
        return json.dumps({"error": f"invalid arguments: {error}"}, ensure_ascii=False)
    try:
        return await session.call_tool(name, arguments)
    except Exception as error:  # noqa: BLE001 - tool failures belong in the conversation
        return json.dumps({"error": f"{type(error).__name__}: {error}"}, ensure_ascii=False)

Malformed JSON arguments happen. When they do, handing the model {"error": "invalid arguments: ..."} gets you a corrected call on the next round, whereas raising kills the session and loses the conversation.

Running the Tool Loop

Now for the agent itself, in src/venice_terminal_agent/agent.py. Start with the system prompt:

SYSTEM_PROMPT = """\
You are a terminal agent. You think with Venice models and act through Apify MCP tools.

Rules:
- Use tools when you need live web data, Actor runs, datasets, or Apify documentation.
- Prefer search-actors and fetch-actor-details before calling an unfamiliar Actor.
- After an Actor run, use get-actor-output when the preview is incomplete.
- Treat tool arguments as untrusted structured data. Do not invent credentials.
- If a tool returns an error, read it and recover instead of guessing.
- If the user declines a tool, continue with what you already know and ask before assuming they want another paid run.
- Keep answers concise and terminal-friendly. Cite Actor names and source URLs when you used them.
"""

DECLINED_TOOL = (
    "user declined to run this Apify tool because it can spend compute; "
    "continue without it or ask them to approve"
)

Every rule there maps to a specific failure we want to avoid. “Prefer search-actors and fetch-actor-details before calling an unfamiliar Actor” exists because a model that guesses at an Actor’s input schema wastes a paid run. The line about declined tools exists because otherwise the model treats a refusal as a transient error and immediately tries again. The Agent class takes the two clients, a model, a round limit, and three callbacks:

class Agent:
    def __init__(
        self,
        venice: VeniceClient,
        apify: ApifyMcp,
        *,
        model: str,
        max_rounds: int,
        on_tool: Callable[[str, str], Awaitable[None] | None] | None = None,
        on_text: Callable[[str], None] | None = None,
        approve_tool: Callable[[str, str], Awaitable[bool] | bool] | None = None,
    ) -> None:
        # each argument is assigned to self, and then:
        self.messages: list[dict[str, Any]] = [
            {"role": "system", "content": SYSTEM_PROMPT},
        ]

Those callbacks are what keep the agent independent of the terminal. on_tool reports a tool call, on_text receives streamed tokens, and approve_tool answers the confirmation question. Swap them out and the same agent works behind a web app or a chat bot. Here is the loop:

    async def ask(
        self,
        question: str,
        *,
        on_text: Callable[[str], None] | None = None,
    ) -> str:
        start = len(self.messages)
        self.messages.append({"role": "user", "content": question})
        tools = self.apify.catalog.openai_tools or None
        text_callback = self._on_text if on_text is None else on_text
        try:
            for _ in range(self.max_rounds):
                message = await self.venice.complete(
                    model=self.model,
                    messages=self.messages,
                    tools=tools,
                    on_text=text_callback,
                )
                calls = list(getattr(message, "tool_calls", None) or [])
                if not calls:
                    content = (message.content or "").strip()
                    self.messages.append(assistant_payload(message))
                    return content or "(empty response)"

                self.messages.append(assistant_payload(message))
                self.messages.extend(await self._execute_calls(calls))

            raise RuntimeError(f"No final answer after {self.max_rounds} tool rounds")
        except BaseException:
            del self.messages[start:]
            raise

That is the entire agent: call the model, and if it asked for tools, run them and call again. The start index and the del in the exception handler are worth a closer look. If a question fails halfway through — network error, Ctrl+C, round limit — the conversation is left with an assistant turn requesting tools that never produced results. Venice will reject the next request, because a tool_calls turn must be followed by matching tool messages. Rolling back to where the question started means a failed question leaves no trace and the REPL stays usable.

Echoing the Assistant Turn

This next function is small and easy to get wrong:

def assistant_payload(message: Any) -> dict[str, Any]:
    """Echo the assistant turn back to Venice without dropping null content.

    Tool-call turns use ``content: null``. ``exclude_none=True`` would strip that
    key and can break the next round. Extra Venice fields such as
    ``reasoning_content`` and ``reasoning_details`` must also be preserved.
    """

    if hasattr(message, "model_dump"):
        payload = message.model_dump(exclude_unset=True)
        payload["role"] = "assistant"
        return payload
    return {"role": "assistant", "content": getattr(message, "content", None)}

The obvious implementation is message.model_dump(exclude_none=True), and it breaks tool calling. A tool-call turn has content: null, and dropping that key changes the shape of the message you send back. exclude_unset=True is the version you want: it keeps null values the model actually set, and omits fields it never sent. It also preserves fields the OpenAI schema does not know about. Reasoning models return reasoning_content and reasoning_details, and those need to survive the round trip so the model keeps its own chain of thought across tool rounds.

Executing and Gating Calls

Models can request several tools in one turn, and there is no reason to run them one at a time. But we do want to ask for approval sequentially, since interleaved confirmation prompts would be unreadable. So we plan first, then execute concurrently:

    async def _execute_calls(self, calls: list[Any]) -> list[dict[str, Any]]:
        planned: list[tuple[Any, str | None]] = []
        for call in calls:
            function = call.function
            name = function.name
            arguments = function.arguments or "{}"
            if self._on_tool is not None:
                maybe = self._on_tool(name, arguments)
                if asyncio.iscoroutine(maybe):
                    await maybe
            if await self._allowed(name, arguments):
                planned.append((call, None))
            else:
                planned.append(
                    (call, json.dumps({"error": DECLINED_TOOL}, ensure_ascii=False))
                )

        async def run(item: tuple[Any, str | None]) -> dict[str, Any]:
            call, denied = item
            if denied is None:
                content = await execute_venice_tool_call(
                    self.apify,
                    name=call.function.name,
                    arguments_json=call.function.arguments,
                )
            else:
                content = denied
            return {
                "role": "tool",
                "tool_call_id": call.id,
                "content": content,
            }

        return list(await asyncio.gather(*[run(item) for item in planned]))

Declined tools still get a tool message. Every tool_call_id needs a reply, and skipping one leaves the conversation malformed. The reply just happens to explain that the user said no. The approval check itself consults both names, since the model works with sanitized names and our allowlist uses MCP names:

    async def _allowed(self, name: str, arguments: str) -> bool:
        mcp_name = self.apify.catalog.mcp_name(name)
        if not (requires_confirmation(name) or requires_confirmation(mcp_name)):
            return True
        if self._approve_tool is None:
            return True
        decision = self._approve_tool(name, arguments)
        if asyncio.iscoroutine(decision):
            return await decision
        return bool(decision)

Adding the CLI

The CLI in src/venice_terminal_agent/cli.py is Typer plus a REPL, and it is the least interesting file in the project — but three details in it are worth copying. The first is that the Typer options are typed as optional and default to None, so the settings loader can tell “not passed” from “passed a falsy value”:

    tools: Annotated[str | None, typer.Option(help="Apify MCP tools query, e.g. actors,docs.")] = None,
    max_rounds: Annotated[int | None, typer.Option(help="Maximum tool-calling rounds per question.")] = None,

Those None defaults are what make the handoff to load_settings() safe, since a flag you did not use never overrides the environment:

        settings = load_settings(
            venice_model=model,
            apify_mcp_transport=transport,
            apify_mcp_tools=tools,
            max_rounds=max_rounds,
            auto_approve_tools=yes or None,
            save_history=save_history or None,
        )

The yes or None is the same idea applied to a boolean flag: --yes sets it, and omitting it passes None rather than False, so AUTO_APPROVE_TOOLS from the environment survives. The second is startup order. Resolve the model, then open the MCP session, then build the agent — and close the Venice client in a finally, since the MCP session and the HTTP clients both need unwinding whether or not the question succeeded:

async def _run(settings: Settings, prompt: str | None) -> None:
    venice = VeniceClient(settings)
    try:
        model_id = await venice.resolve_model()
        async with ApifyMcpSession(settings) as apify:
            agent = Agent(
                venice,
                apify,
                model=model_id,
                max_rounds=settings.max_rounds,
                on_tool=print_tool_call,
                approve_tool=_make_approver(settings.auto_approve_tools),
            )
            print_banner(model_id, apify.catalog.names(), anonymous=apify.anonymous)
            if prompt:
                await _ask(agent, prompt)
                return
            await _repl(agent, save_history=settings.save_history)
    finally:
        await venice.aclose()

The third is the approver, which is the one piece of the agent that exists purely to protect your Apify bill:

def _make_approver(auto_approve: bool):
    def approve(name: str, arguments: str) -> bool:
        if auto_approve:
            return True
        if not sys.stdin.isatty():
            print_info(f"Skipped {name}: paid Apify tools need a TTY or --yes.")
            return False
        preview = arguments if len(arguments) <= 240 else f"{arguments[:240]}…"
        try:
            return Confirm.ask(
                f"Run Apify tool [bold]{name}[/bold]({preview})? This can spend compute",
                default=False,
            )
        except (KeyboardInterrupt, EOFError):
            return False

    return approve

The isatty() check is the part people forget. Run the agent from cron or CI and there is nobody to answer the prompt, so a naive implementation either hangs forever or silently approves. Here it declines, says why, and lets the model carry on with the read-only tools. default=False means a stray Enter does not start a paid run, and interrupting the prompt counts as a no. The rest of the module is ordinary terminal work, so it is worth knowing what is there rather than reading it: a prompt_toolkit REPL loop, a _handle_command() lookup for the slash commands, a render.py of Rich helpers, and a _settings_error() that turns a missing VENICE_API_KEY into a readable message instead of a Pydantic traceback. Three of those carry a decision:

Piece	Decision worth keeping
REPL history	In memory unless you pass `--save-history`. For an agent you may ask about private material, writing every prompt to `~/.local/share` by default is a poor choice.
`Ctrl+C` handling	Cancels the question and returns to the prompt instead of exiting, which pairs with the history rollback in `Agent.ask()` so a cancelled question leaves a clean conversation.
`StreamPrinter`	Prints streamed tokens with `markup=False` and `highlight=False`, or Rich reads square brackets in model output as its own formatting tags.

The slash commands are /help, /clear, /quit, and two that earn their keep: /tools prints the loaded catalog, which usually explains why the agent picked an odd tool, and /reload picks up Actors you added to your Apify account mid-session. Finally, wire the entry point up in pyproject.toml so uv run venice-agent works:

[project.scripts]
venice-agent = "venice_terminal_agent.cli:main"

Running the Agent

Start an interactive session:

uv run venice-agent

Or ask one question and exit:

uv run venice-agent "Find an Apify Actor that scrapes Hacker News and summarize how to call it"

You’ll see the banner, then the tool calls as they happen:

╭──────────────────────────────────────────────╮
│ Venice terminal agent                        │
│ Model: zai-org-glm-5-2                       │
│ Apify: Apify authenticated                   │
│ Tools: search-actors, fetch-actor-details, … │
╰──────────────────────────────────────────────╯

→ search-actors({"search": "hacker news scraper", "limit": 5})
→ fetch-actor-details({"actor": "epctex/hackernews-scraper"})

epctex/hackernews-scraper takes a `startUrls` array and a `maxItems` cap…

Read the Apify line on that banner before anything else. If it says “anonymous Apify tools only”, your APIFY_TOKEN did not load, and that is much better to notice now than after ten minutes of wondering why the agent refuses to run an Actor. Restrict the tool catalog when you know what you need:

uv run venice-agent --tools actors,docs,apify/rag-web-browser

A smaller catalog is not just about cost. Models generally pick better when there are fewer, more relevant tools to choose between, and --tools is the cheapest way to narrow the choice. Run the MCP server locally instead of using the hosted one:

uv run venice-agent --transport stdio

This one needs Node.js on your PATH, since it launches @apify/actors-mcp-server through npx, and it needs an APIFY_TOKEN — there is no anonymous mode for the local server. And when you genuinely want unattended Actor runs:

uv run venice-agent --yes "Crawl https://docs.venice.ai and list the guides about tool calling"

Testing the Pieces

None of the interesting logic here needs a network. A FakeVenice that pops from a scripted list of replies, plus a FakeApify that builds a real ToolCatalog from SimpleNamespace tools, is enough to drive a full tool round:

class FakeVenice:
    def __init__(self, replies: list[object]) -> None:
        self.replies = list(replies)

    async def complete(self, **kwargs: object) -> object:
        return self.replies.pop(0)


@pytest.mark.asyncio
async def test_agent_runs_tool_then_answers() -> None:
    venice = FakeVenice(
        [
            _tool_message("search-actors", '{"query": "hacker news"}'),
            _text_message("Use apify/rag-web-browser."),
        ]
    )
    apify = FakeApify()
    agent = Agent(venice, apify, model="test-model", max_rounds=4)

    answer = await agent.ask("Find a scraper for Hacker News")

    assert apify.calls == [("search-actors", {"query": "hacker news"})]
    roles = [message["role"] for message in agent.messages]
    assert roles == ["system", "user", "assistant", "tool", "assistant"]

Asserting on the sequence of roles is a good habit for agent code. It catches the malformed-conversation bugs that are otherwise invisible until Venice returns a 400. Three more tests are worth writing, and all of them assert on agent.messages in the same way. That a failed run rolls history back to just ["system"], whether it failed on a Venice error or by exhausting max_rounds. That a read-only tool still runs when the approver returns False. And that a declined paid tool leaves a tool message containing declined while apify.calls stays empty. Run the suite with:

uv run pytest

Privacy and Cost Notes

An agent that reaches two APIs is worth being precise about:

Layer	What sees the data
Local CLI	Your question, configuration, conversation history, and tool results stay in memory on your machine
Venice chat completions	System prompt, your questions, tool schemas, and tool results are sent to Venice, which does not retain them
Apify MCP	Tool arguments the model generates, such as search terms and target URLs
Apify Actors	The sites an approved Actor visits, plus whatever the Actor stores in your Apify datasets
Local disk	Nothing, unless you pass `--save-history`

Venice’s zero data retention covers the model side. It does not cover Apify, and an Actor run writes results into your Apify account. If that matters for a particular task, run without APIFY_TOKEN and stick to the anonymous discovery tools. On cost, three habits go a long way:

Leave --yes off during development. Watching which Actors the model wants to run is informative in itself.
Use --tools to narrow the catalog to Actors you have actually reviewed.
Keep max_rounds modest. Twelve rounds is plenty for research tasks, and a lower ceiling caps the damage when a model gets stuck in a loop.

Extending This Example

The loop is the foundation. Once it works, useful directions include:

Add a second MCP server. Nothing in Agent is Apify-specific, so merging catalogs from several servers mostly means namespacing tool names.
Persist conversations to SQLite so you can resume a session or audit what an Actor returned.
Add per-tool budgets that track Actor runs and stop at a ceiling, rather than confirming each one.
Cache tool results by name and arguments, so repeated documentation lookups do not re-crawl.
Pin a model with --model and compare tool-selection quality against function_calling_default.
Swap the approver for a policy function that auto-approves specific Actors with specific arguments and prompts for everything else.

For a smaller starting point without MCP, Building a Tool-Using Agent covers the same loop with three local Python functions.

Finishing Up

Thanks for reading! Hopefully this helped you build a terminal agent that thinks with Venice and acts through Apify. The pattern worth taking away is how little of this code is about intelligence. The model returns tool calls, and your code decides which ones are allowed to run, how their results come back, and what happens when something fails. Once those decisions are explicit, adding capability is mostly a matter of pointing the agent at more tools.

Learn

Building a Terminal Agent with Apify

What We’re Building

Setting Up the Project

Loading Configuration

Talking to Venice

Discovering a Model at Runtime

Streaming Completions

Converting MCP Tools into Venice Tools

Formatting Results Back into Context

Marking the Tools That Cost Money

Connecting to Apify over MCP

Owning the Transport

Executing a Tool Call

Running the Tool Loop

Echoing the Assistant Turn

Executing and Gating Calls

Adding the CLI

Running the Agent

Testing the Pieces

Privacy and Cost Notes

Extending This Example

Finishing Up

​What We’re Building

​Setting Up the Project

​Loading Configuration

​Talking to Venice

​Discovering a Model at Runtime

​Streaming Completions

​Converting MCP Tools into Venice Tools

​Formatting Results Back into Context

​Marking the Tools That Cost Money

​Connecting to Apify over MCP

​Owning the Transport

​Executing a Tool Call

​Running the Tool Loop

​Echoing the Assistant Turn

​Executing and Gating Calls

​Adding the CLI

​Running the Agent

​Testing the Pieces

​Privacy and Cost Notes

​Extending This Example

​Finishing Up

What We’re Building

Setting Up the Project

Loading Configuration

Talking to Venice

Discovering a Model at Runtime

Streaming Completions

Converting MCP Tools into Venice Tools

Formatting Results Back into Context

Marking the Tools That Cost Money

Connecting to Apify over MCP

Owning the Transport

Executing a Tool Call

Running the Tool Loop

Echoing the Assistant Turn

Executing and Gating Calls

Adding the CLI

Running the Agent

Testing the Pieces

Privacy and Cost Notes

Extending This Example

Finishing Up