Framework integrations

Google ADK Integration

Plug MuBit into ADK's Runner as a BaseMemoryService. Session events auto-ingest; memory search enriches agent context.

The mubit-adk adapter implements ADK's BaseMemoryService. Pass it to Runner(memory_service=…) and every session event is automatically ingested into MuBit; memory search enriches agent context on the next turn.

pip install mubit-adk[adk]

Minimal usage

google_adk_minimal.py

from mubit_adk import MubitMemoryService
from google.adk.agents import LlmAgent, SequentialAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
 
memory_service = MubitMemoryService(api_key="mbt_...")
 
agent = SequentialAgent(name="coordinator", sub_agents=[flight_agent, hotel_agent, planner_agent])
runner = Runner(
    agent=agent,
    app_name="travel",
    session_service=InMemorySessionService(),
    memory_service=memory_service,
)
 
# MAS extensions
await memory_service.checkpoint(app_name="travel", user_id="user-1", snapshot="Plan complete")
await memory_service.register_agent(user_id="user-1", agent_id="planner", role="itinerary")

Full example: Travel planner

A SequentialAgent with Gemini, tool calling, and MAS coordination. Three agents (flight finder, hotel finder, itinerary planner) collaborate through MuBit memory.

travel_planner/main.py

from google.adk.agents import LlmAgent, SequentialAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from mubit_adk import MubitMemoryService
 
mubit_memory = MubitMemoryService(endpoint="https://api.mubit.ai", api_key="mbt_...")
 
flight_finder = LlmAgent(
    name="flight_finder", model="gemini-3.1-flash-lite-preview",
    description="Finds the best flights",
    instruction="Search for flights and recommend the best option.",
    tools=[search_flights], output_key="flight_results",
)
hotel_finder = LlmAgent(
    name="hotel_finder", model="gemini-3.1-flash-lite-preview",
    description="Finds accommodations",
    instruction="Search for hotels and recommend the best option.",
    tools=[search_hotels], output_key="hotel_results",
)
itinerary_planner = LlmAgent(
    name="itinerary_planner", model="gemini-3.1-flash-lite-preview",
    description="Creates day-by-day itinerary",
    instruction="Combine flight and hotel info into a complete travel plan.",
    output_key="final_itinerary",
)
 
coordinator = SequentialAgent(
    name="travel_coordinator",
    sub_agents=[flight_finder, hotel_finder, itinerary_planner],
)
 
runner = Runner(
    agent=coordinator, app_name="travel",
    session_service=InMemorySessionService(),
    memory_service=mubit_memory,
)
 
# After the pipeline runs, record MuBit operations
await mubit_memory.register_agent(user_id="user-1", agent_id="flight_finder", role="finder")
await mubit_memory.checkpoint(app_name="travel", user_id="user-1", snapshot="Plan complete")
await mubit_memory.record_outcome(
    app_name="travel", user_id="user-1", reference_id="travel-001",
    outcome="success", rationale="Itinerary confirmed by user.",
)

Extended MuBit features

MubitMemoryService is async — every method returns a coroutine.

await memory_service.register_agent(user_id="...", agent_id="...", role="...", read_scopes=[...], write_scopes=[...])
await memory_service.handoff(user_id="...", from_agent_id="...", to_agent_id="...", task_id="...", content="...", requested_action="review")
await memory_service.feedback(user_id="...", handoff_id="...", verdict="approve", from_agent_id="...")
await memory_service.checkpoint(app_name="...", user_id="...", snapshot="...", label="...")
await memory_service.record_outcome(app_name="...", user_id="...", reference_id="...", outcome="success", rationale="...")
await memory_service.surface_strategies(session_id="...")
await memory_service.diagnose(user_id="...", error_text="...")
await memory_service.archive(app_name="...", user_id="...", content="...", artifact_kind="...")
await memory_service.dereference(user_id="...", reference_id="...")

Gotchas

app_name + user_id form the namespace ADK uses internally; the adapter joins them into a MuBit session id.
Auto-ingestion is opt-out, not opt-in. Set auto_ingest_events=False on MubitMemoryService(...) if you want to capture only what you call manually.
Gemini streaming responses are captured incrementally; the assembled output is what gets stored.

Version compatibility

`mubit-adk`	`mubit-sdk`	`google-adk`
`0.5.x`	`>= 0.6.0`	`>= 1.0.0`