Fix/markdown stream improvements, uvloop (#582)

* typesafe tests

* lint

* enhancements round one

* markdown improvements

* anyio/uvloop/streaming improvements*lots

* lint

* algorithm tweaks

Author: evalstate

pyproject.toml

@@ -39,7 +39,8 @@ dependencies = [
     "keyring>=24.3.1",
     "python-frontmatter>=1.1.0",
     "agent-client-protocol>=0.7.0",
-    "tiktoken>=0.12.0"
+    "tiktoken>=0.12.0",
+    "uvloop>=0.22.1",
 ]
 
 [project.optional-dependencies]

src/fast_agent/acp/slash_commands.py

@@ -558,7 +558,15 @@ async def _handle_status_system(self) -> str:
         if error:
             return error
 
-        system_prompt = agent.instruction if isinstance(agent, InstructionAwareAgent) else None
+        agent_name = (
+            agent.name if isinstance(agent, InstructionAwareAgent) else self.current_agent_name
+        )
+
+        system_prompt = None
+        if agent_name in self._session_instructions:
+            system_prompt = self._session_instructions[agent_name]
+        elif isinstance(agent, InstructionAwareAgent):
+            system_prompt = agent.instruction
         if not system_prompt:
             return "\n".join(
                 [
@@ -569,9 +577,6 @@ async def _handle_status_system(self) -> str:
             )
 
         # Format the response
-        agent_name = (
-            agent.name if isinstance(agent, InstructionAwareAgent) else self.current_agent_name
-        )
         lines = [
             heading,
             "",

src/fast_agent/cli/__main__.py

@@ -4,14 +4,21 @@
 
 from fast_agent.cli.constants import GO_SPECIFIC_OPTIONS, KNOWN_SUBCOMMANDS
 from fast_agent.cli.main import app
+from fast_agent.utils.async_utils import configure_uvloop, ensure_event_loop
 
 # if the arguments would work with "go" we'll just route to it
 
 
 def main():
     """Main entry point that handles auto-routing to 'go' command."""
+    requested_uvloop, enabled_uvloop = configure_uvloop()
+    if requested_uvloop and not enabled_uvloop:
+        print(
+            "FAST_AGENT_UVLOOP is set but uvloop is unavailable; falling back to asyncio.",
+            file=sys.stderr,
+        )
     try:
-        loop = asyncio.get_event_loop()
+        loop = ensure_event_loop()
 
         def _log_asyncio_exception(loop: asyncio.AbstractEventLoop, context: dict) -> None:
             import logging

src/fast_agent/cli/commands/README.md

@@ -118,3 +118,7 @@ fast-agent serve --description "Interact with the {agent} workflow via MCP"
 # Use per-connection instances to isolate history between clients
 fast-agent serve --instance-scope=connection --transport=http
 ```
+
+### Environment toggles
+
+- uvloop is enabled by default when installed (non-Windows); set `FAST_AGENT_DISABLE_UV_LOOP=1` to opt out.

src/fast_agent/cli/commands/auth.py

@@ -16,6 +16,7 @@
     list_keyring_tokens,
 )
 from fast_agent.ui.console import console
+from fast_agent.utils.async_utils import run_sync
 
 app = typer.Typer(help="Manage OAuth authentication state for MCP servers")
 
@@ -395,9 +396,7 @@ async def _run_login():
             typer.echo(f"Login failed: {e}")
             return False
 
-    import asyncio
-
-    ok = asyncio.run(_run_login())
+    ok = bool(run_sync(_run_login))
     if ok:
         from fast_agent.mcp.oauth_client import compute_server_identity
 

src/fast_agent/cli/commands/go.py

@@ -9,12 +9,10 @@
 
 import typer
 
-from fast_agent import FastAgent
-from fast_agent.agents.llm_agent import LlmAgent
 from fast_agent.cli.commands.server_helpers import add_servers_to_config, generate_server_name
 from fast_agent.cli.commands.url_parser import generate_server_configs, parse_server_urls
 from fast_agent.constants import DEFAULT_AGENT_INSTRUCTION
-from fast_agent.ui.console_display import ConsoleDisplay
+from fast_agent.utils.async_utils import configure_uvloop, create_event_loop, ensure_event_loop
 
 app = typer.Typer(
     help="Run an interactive agent directly from the command line without creating an agent.py file",
@@ -126,7 +124,10 @@ async def _run_agent(
     permissions_enabled: bool = True,
 ) -> None:
     """Async implementation to run an interactive agent."""
+    from fast_agent import FastAgent
+    from fast_agent.agents.llm_agent import LlmAgent
     from fast_agent.mcp.prompts.prompt_load import load_prompt
+    from fast_agent.ui.console_display import ConsoleDisplay
 
     # Create the FastAgent instance
 
@@ -273,6 +274,7 @@ def run_async_agent(
     permissions_enabled: bool = True,
 ):
     """Run the async agent function with proper loop handling."""
+    configure_uvloop()
     server_list = servers.split(",") if servers else None
 
     # Parse URLs and generate server configurations if provided
@@ -346,19 +348,12 @@ def run_async_agent(
                 continue
 
     # Check if we're already in an event loop
-    try:
-        loop = asyncio.get_event_loop()
-        if loop.is_running():
-            # We're inside a running event loop, so we can't use asyncio.run
-            # Instead, create a new loop
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-        _set_asyncio_exception_handler(loop)
-    except RuntimeError:
-        # No event loop exists, so we'll create one
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-        _set_asyncio_exception_handler(loop)
+    loop = ensure_event_loop()
+    if loop.is_running():
+        # We're inside a running event loop, so we can't use asyncio.run
+        # Instead, create a new loop
+        loop = create_event_loop()
+    _set_asyncio_exception_handler(loop)
 
     try:
         loop.run_until_complete(

src/fast_agent/cli/main.py

@@ -1,26 +1,51 @@
 """Main CLI entry point for MCP Agent."""
 
+import importlib
+
+import click
 import typer
-from rich.table import Table
+import typer.main
+from typer.core import TyperGroup
 
-from fast_agent.cli.commands import acp, auth, check_config, go, quickstart, serve, setup
 from fast_agent.cli.terminal import Application
 from fast_agent.ui.console import console as shared_console
 
+LAZY_SUBCOMMANDS: dict[str, str] = {
+    "go": "fast_agent.cli.commands.go:app",
+    "serve": "fast_agent.cli.commands.serve:app",
+    "acp": "fast_agent.cli.commands.acp:app",
+    "setup": "fast_agent.cli.commands.setup:app",
+    "check": "fast_agent.cli.commands.check_config:app",
+    "auth": "fast_agent.cli.commands.auth:app",
+    "quickstart": "fast_agent.cli.commands.quickstart:app",
+    "bootstrap": "fast_agent.cli.commands.quickstart:app",
+}
+
+
+class LazyGroup(TyperGroup):
+    lazy_subcommands: dict[str, str] = {}
+
+    def list_commands(self, ctx: click.Context) -> list[str]:
+        return sorted(self.lazy_subcommands)
+
+    def get_command(self, ctx: click.Context, cmd_name: str) -> click.Command | None:
+        target = self.lazy_subcommands.get(cmd_name)
+        if not target:
+            return None
+        module_path, app_name = target.split(":", 1)
+        module = importlib.import_module(module_path)
+        typer_app = getattr(module, app_name)
+        command = typer.main.get_command(typer_app)
+        command.name = cmd_name
+        return command
+
+
 app = typer.Typer(
+    cls=LazyGroup,
     help="Use `fast-agent go --help` for interactive shell arguments and options.",
     add_completion=False,  # We'll add this later when we have more commands
 )
-
-# Subcommands
-app.add_typer(go.app, name="go", help="Run an interactive agent directly from the command line")
-app.add_typer(serve.app, name="serve", help="Run FastAgent as an MCP server")
-app.add_typer(acp.app, name="acp", help="Run FastAgent as an ACP stdio server")
-app.add_typer(setup.app, name="setup", help="Set up a new agent project")
-app.add_typer(check_config.app, name="check", help="Show or diagnose fast-agent configuration")
-app.add_typer(auth.app, name="auth", help="Manage OAuth authentication for MCP servers")
-app.add_typer(quickstart.app, name="bootstrap", help="Create example applications")
-app.add_typer(quickstart.app, name="quickstart", help="Create example applications")
+LazyGroup.lazy_subcommands = LAZY_SUBCOMMANDS
 
 # Shared application context
 application = Application()
@@ -32,6 +57,7 @@ def show_welcome() -> None:
     """Show a welcome message with available commands, using new styling."""
     from importlib.metadata import version
 
+    from rich.table import Table
     from rich.text import Text
 
     try:

src/fast_agent/context.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-import asyncio
-import concurrent.futures
 import logging
 import uuid
 from os import PathLike
@@ -27,6 +25,7 @@
 from fast_agent.core.logging.transport import create_transport
 from fast_agent.mcp_server_registry import ServerRegistry
 from fast_agent.skills import SkillRegistry
+from fast_agent.utils.async_utils import run_sync
 
 if TYPE_CHECKING:
     from fast_agent.acp.acp_context import ACPContext
@@ -253,22 +252,10 @@ def get_current_context() -> Context:
     """
     global _global_context
     if _global_context is None:
-        try:
-            # Try to get the current event loop
-            loop = asyncio.get_event_loop()
-            if loop.is_running():
-                # Create a new loop in a separate thread
-                def run_async():
-                    new_loop = asyncio.new_event_loop()
-                    asyncio.set_event_loop(new_loop)
-                    return new_loop.run_until_complete(initialize_context())
-
-                with concurrent.futures.ThreadPoolExecutor() as pool:
-                    _global_context = pool.submit(run_async).result()
-            else:
-                _global_context = loop.run_until_complete(initialize_context())
-        except RuntimeError:
-            _global_context = asyncio.run(initialize_context())
+        result = run_sync(initialize_context)
+        if result is None:
+            raise RuntimeError("Failed to initialize global context")
+        _global_context = result
     return _global_context
 
 

src/fast_agent/core/executor/workflow_signal.py

@@ -143,8 +143,8 @@ async def wait_for_signal(self, signal, timeout_seconds=None):
         if timeout_seconds:
             print(f"(Timeout in {timeout_seconds} seconds)")
 
-        # Use asyncio.get_event_loop().run_in_executor to make input non-blocking
-        loop = asyncio.get_event_loop()
+        # Use asyncio.get_running_loop().run_in_executor to make input non-blocking
+        loop = asyncio.get_running_loop()
         if timeout_seconds is not None:
             try:
                 value = await asyncio.wait_for(

src/fast_agent/core/logging/logger.py

@@ -21,6 +21,7 @@
     ProgressListener,
 )
 from fast_agent.core.logging.transport import AsyncEventBus, EventTransport
+from fast_agent.utils.async_utils import ensure_event_loop
 
 
 class Logger:
@@ -34,19 +35,9 @@ def __init__(self, namespace: str) -> None:
         self.namespace = namespace
         self.event_bus = AsyncEventBus.get()
 
-    def _ensure_event_loop(self):
-        """Ensure we have an event loop we can use."""
-        try:
-            return asyncio.get_running_loop()
-        except RuntimeError:
-            # If no loop is running, create a new one
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-            return loop
-
     def _emit_event(self, event: Event) -> None:
         """Emit an event by running it in the event loop."""
-        loop = self._ensure_event_loop()
+        loop = ensure_event_loop()
         if loop.is_running():
             # If we're in a thread with a running loop, schedule the coroutine
             asyncio.create_task(self.event_bus.emit(event))