Loading the notebook…
Loading the notebook…
Notebooks · Specialist agents · 47
End-to-end workflow:
Its imports and arguments are checked against shipit-agent 3.0. The outputs were saved when it was written, for an earlier release.
from pathlib import Path
def _find_notebooks_dir() -> Path:
cwd = Path.cwd().resolve()
if cwd.name == 'notebooks':
return cwd
candidate = cwd / 'notebooks'
return candidate if candidate.is_dir() else cwd
WORKSPACE = _find_notebooks_dir() / '_pm_workspace'
WORKSPACE.mkdir(parents=True, exist_ok=True)
print('workspace:', WORKSPACE)
Defaults to SimpleEchoLLM so this notebook runs clean without creds. Uncomment one of the Bedrock / LiteLLM / LiteLLM-proxy blocks to wire in a real model.
# from shipit_agent.llms import build_llm_from_settings
# --- Option A: AWS Bedrock ------------------------------------
# llm = build_llm_from_settings({
# 'provider': 'bedrock',
# 'model': 'bedrock/anthropic.claude-sonnet-4-5-v2:0',
# }, provider='bedrock')
# --- Option B: LiteLLM direct ---------------------------------
# llm = build_llm_from_settings({
# 'provider': 'litellm',
# 'model': 'openai/gpt-4o-mini',
# }, provider='litellm')
# --- Option C: LiteLLM proxy ----------------------------------
# from shipit_agent.llms import LiteLLMProxyChatLLM
# llm = LiteLLMProxyChatLLM(
# model='gpt-4o-mini',
# api_base='https://litellm.my-company.internal',
# api_key='sk-proxy-token',
# )
# Zero-credential default — deterministic echo model.
from shipit_agent.llms import SimpleEchoLLM
llm = SimpleEchoLLM()
print('llm:', type(llm).__name__)
Both GitHubTool and SlackTool are HTTPConnectorToolBase subclasses — they go through _request_json(record=..., method=..., path=..., query=..., body=...). We replace that method with a canned-response function so the workflow runs end-to-end with zero external API traffic.
from shipit_agent.integrations import CredentialRecord, InMemoryCredentialStore
from shipit_agent.tools.github import GitHubTool
from shipit_agent.tools.slack import SlackTool
store = InMemoryCredentialStore()
store.set(CredentialRecord(
key='github', provider='github',
secrets={'token': 'ghp_demo'},
metadata={'base_url': 'https://api.github.com', 'auth_scheme': 'Bearer'},
))
store.set(CredentialRecord(
key='slack', provider='slack',
secrets={'token': 'xoxb-demo'},
))
gh = GitHubTool(credential_store=store)
slack = SlackTool(credential_store=store)
print('tools ready:', gh.name, slack.name)
# Canned GitHub responses for /repos/<owner>/<repo>/pulls with state=closed.
_CANNED_PRS = {
'acme/api': [
{'number': 412, 'title': 'Fix rate-limit edge case on /search',
'state': 'closed', 'merged_at': '2026-04-22T10:15:00Z',
'user': {'login': 'adeleke'},
'html_url': 'https://github.com/acme/api/pull/412'},
{'number': 414, 'title': 'Add pagination cursor to /v2/orders',
'state': 'closed', 'merged_at': '2026-04-23T09:02:00Z',
'user': {'login': 'hmarta'},
'html_url': 'https://github.com/acme/api/pull/414'},
],
'acme/web': [
{'number': 289, 'title': 'Dashboard sidebar redesign',
'state': 'closed', 'merged_at': '2026-04-21T14:40:00Z',
'user': {'login': 'rahul'},
'html_url': 'https://github.com/acme/web/pull/289'},
{'number': 291, 'title': 'Billing page empty-state copy tweak',
'state': 'closed', 'merged_at': '2026-04-22T16:05:00Z',
'user': {'login': 'kpatel'},
'html_url': 'https://github.com/acme/web/pull/291'},
],
'acme/mobile': [
{'number': 178, 'title': 'Android push notifications crash fix',
'state': 'closed', 'merged_at': '2026-04-23T11:30:00Z',
'user': {'login': 'shinji'},
'html_url': 'https://github.com/acme/mobile/pull/178'},
],
}
def _fake_github(*, record, method, path, query=None, body=None):
# Demo only — return canned PR lists.
for slug, prs in _CANNED_PRS.items():
if path == f'/repos/{slug}/pulls':
return prs
return []
gh._request_json = _fake_github
def _fake_slack(*, record, method, path, query=None, body=None):
# Demo only — return Slack-like {ok: True} payloads.
if path == '/chat.postMessage':
return {'ok': True, 'channel': body.get('channel'),
'ts': '1714000000.000100', 'message': {'text': body.get('text', '')}}
return {'ok': True}
slack._request_json = _fake_slack
print('stubs installed')
Each tool call goes through the same ToolContext path the Agent uses — so the round-trip mirrors what happens when an LLM emits a tool call.
from shipit_agent.tools.base import ToolContext
ctx = ToolContext(prompt='pm digest', state={'credential_store': store})
REPOS = [('acme', 'api'), ('acme', 'web'), ('acme', 'mobile')]
digest: dict[str, list[dict]] = {}
for owner, repo in REPOS:
out = gh.run(ctx, action='list_pulls', owner=owner, repo=repo, state='closed')
items = out.metadata.get('items') or []
digest[f'{owner}/{repo}'] = items
print(f'{owner}/{repo}: {len(items)} merged PRs')
The DashboardRenderTool emits a self-contained HTML document we can ship as a PR-review-ready artifact or attach to the Slack post.
from shipit_agent.tools.dashboard_render import DashboardRenderTool
dash = DashboardRenderTool(workspace_root=WORKSPACE)
metric_items = [
{'label': slug, 'value': str(len(prs)), 'sub': 'merged'}
for slug, prs in digest.items()
]
total = sum(len(v) for v in digest.values())
metric_items.append({'label': 'Total', 'value': str(total), 'sub': 'this week'})
timeline = []
for slug, prs in digest.items():
for pr in prs:
timeline.append({
'period': pr['merged_at'][:10],
'head': f"{slug} #{pr['number']}",
'desc': pr['title'],
'dot_color': '#185fa5',
'tags': [{'text': pr.get('user', {}).get('login', '?'), 'color': 'blue'}],
})
timeline.sort(key=lambda t: t['period'])
result = dash.run(
ToolContext(prompt='pm digest', state={'artifact_workspace_root': str(WORKSPACE)}),
title='Weekly PR Digest — Acme Platform',
subtitle='Week of 2026-04-20',
lang='en',
sections=[
{'type': 'metrics', 'title': 'Merges by repo',
'columns': len(metric_items), 'items': metric_items},
{'type': 'timeline', 'title': 'Merge timeline', 'items': timeline},
{'type': 'verdict', 'title': 'PM takeaway',
'text': (f'**{total} PRs merged this week** across '
f'{len(REPOS)} repos. Biggest driver: API '
f'pagination + rate-limiting work.')},
],
export=True,
)
print(result.text)
print('artifact path:', result.metadata.get('path'))
We post a plain-text summary of the digest to #product-updates. In a real run you'd also upload the HTML dashboard as a file attachment — here we just send a text message.
text_lines = ['*Weekly PR Digest — week of 2026-04-20*', '']
for slug, prs in digest.items():
text_lines.append(f'*{slug}* — {len(prs)} merged')
for pr in prs:
author = pr.get('user', {}).get('login', '?')
text_lines.append(f' • #{pr["number"]} {pr["title"]} (@{author})')
text_lines.append('')
text = '\n'.join(text_lines)
slack_out = slack.run(ctx, action='post_message',
channel='C_PRODUCT_UPDATES', text=text)
print(slack_out.text)
Drop the same tools into Autopilot(..., artifacts=True) and the rendered dashboard flows through ArtifactCollector.ingest_tool_metadata automatically — no glue code.
from shipit_agent.autopilot import ArtifactCollector
collector = ArtifactCollector()
collector.ingest_tool_metadata(result.metadata, iteration=1)
for a in collector.all():
print(f'{a.kind:<6} {a.name:<30} {len(a.content):>6,} chars')
GitHubTool credential (personal access token) to run this on your actual repos.docs-app/content/source/tools/github.md for the full action surface.Autopilot(...) run so the digest happens on a daily schedule via the scheduler daemon (notebook 39).